1
//! Abstract code to manage a set of circuits.
2
//!
3
//! This module implements the real logic for deciding when and how to
4
//! launch circuits, and for which circuits to hand out in response to
5
//! which requests.
6
//!
7
//! For testing and abstraction purposes, this module _does not_
8
//! actually know anything about circuits _per se_.  Instead,
9
//! everything is handled using a set of traits that are internal to this
10
//! crate:
11
//!
12
//!  * [`AbstractCirc`] is a view of a circuit.
13
//!  * [`AbstractCircBuilder`] knows how to build an `AbstractCirc`.
14
//!
15
//! Using these traits, the [`AbstractCircMgr`] object manages a set of
16
//! circuits, launching them as necessary, and keeping track of the
17
//! restrictions on their use.
18

            
19
// TODO:
20
// - Testing
21
//    - Error from prepare_action()
22
//    - Error reported by restrict_mut?
23

            
24
use crate::config::CircuitTiming;
25
use crate::usage::{SupportedCircUsage, TargetCircUsage};
26
use crate::{timeouts, DirInfo, Error, PathConfig, Result};
27

            
28
use retry_error::RetryError;
29
use tor_async_utils::mpsc_channel_no_memquota;
30
use tor_basic_utils::retry::RetryDelay;
31
use tor_config::MutCfg;
32
use tor_error::{debug_report, info_report, internal, warn_report, AbsRetryTime, HasRetryTime};
33
#[cfg(feature = "vanguards")]
34
use tor_guardmgr::vanguards::VanguardMgr;
35
use tor_linkspec::CircTarget;
36
use tor_proto::circuit::{CircParameters, Path, UniqId};
37
use tor_rtcompat::{Runtime, SleepProviderExt};
38

            
39
use async_trait::async_trait;
40
use futures::channel::mpsc;
41
use futures::future::{FutureExt, Shared};
42
use futures::stream::{FuturesUnordered, StreamExt};
43
use futures::task::SpawnExt;
44
use oneshot_fused_workaround as oneshot;
45
use std::collections::HashMap;
46
use std::fmt::Debug;
47
use std::hash::Hash;
48
use std::panic::AssertUnwindSafe;
49
use std::sync::{self, Arc, Weak};
50
use std::time::{Duration, Instant};
51
use tracing::{debug, warn};
52
use weak_table::PtrWeakHashSet;
53

            
54
mod streams;
55

            
56
/// Description of how we got a circuit.
57
#[non_exhaustive]
58
#[derive(Debug, Copy, Clone, Eq, PartialEq)]
59
pub(crate) enum CircProvenance {
60
    /// This channel was newly launched, or was in progress and finished while
61
    /// we were waiting.
62
    NewlyCreated,
63
    /// This channel already existed when we asked for it.
64
    Preexisting,
65
}
66

            
67
#[derive(Clone, Debug, thiserror::Error)]
68
#[non_exhaustive]
69
pub enum RestrictionFailed {
70
    /// Tried to restrict a specification, but the circuit didn't support the
71
    /// requested usage.
72
    #[error("Specification did not support desired usage")]
73
    NotSupported,
74
}
75

            
76
/// Minimal abstract view of a circuit.
77
///
78
/// From this module's point of view, circuits are simply objects
79
/// with unique identities, and a possible closed-state.
80
#[async_trait]
81
pub(crate) trait AbstractCirc: Debug {
82
    /// Type for a unique identifier for circuits.
83
    type Id: Clone + Debug + Hash + Eq + Send + Sync;
84
    /// Return the unique identifier for this circuit.
85
    ///
86
    /// # Requirements
87
    ///
88
    /// The values returned by this function are unique for distinct
89
    /// circuits.
90
    fn id(&self) -> Self::Id;
91

            
92
    /// Return true if this circuit is usable for some purpose.
93
    ///
94
    /// Reasons a circuit might be unusable include being closed.
95
    fn usable(&self) -> bool;
96

            
97
    /// Return a [`Path`] object describing all the hops in this circuit.
98
    ///
99
    /// Returns an error if the circuit is closed.
100
    ///
101
    /// Note that this `Path` is not automatically updated if the circuit is
102
    /// extended.
103
    fn path_ref(&self) -> tor_proto::Result<Arc<Path>>;
104

            
105
    /// Return the number of hops in this circuit.
106
    ///
107
    /// Returns an error if the circuit is closed.
108
    ///
109
    /// NOTE: This function will currently return only the number of hops
110
    /// _currently_ in the circuit. If there is an extend operation in progress,
111
    /// the currently pending hop may or may not be counted, depending on whether
112
    /// the extend operation finishes before this call is done.
113
    fn n_hops(&self) -> tor_proto::Result<usize>;
114

            
115
    /// Return true if this circuit is closed and therefore unusable.
116
    fn is_closing(&self) -> bool;
117

            
118
    /// Return a process-unique identifier for this circuit.
119
    fn unique_id(&self) -> UniqId;
120

            
121
    /// Extend the circuit via the most appropriate handshake to a new `target` hop.
122
    async fn extend<T: CircTarget + Sync>(
123
        &self,
124
        target: &T,
125
        params: CircParameters,
126
    ) -> tor_proto::Result<()>;
127
}
128

            
129
/// A plan for an `AbstractCircBuilder` that can maybe be mutated by tests.
130
///
131
/// You should implement this trait using all default methods for all code that isn't test code.
132
pub(crate) trait MockablePlan {
133
    /// Add a reason string that was passed to `SleepProvider::block_advance()` to this object
134
    /// so that it knows what to pass to `::release_advance()`.
135
    fn add_blocked_advance_reason(&mut self, _reason: String) {}
136
}
137

            
138
/// An object that knows how to build circuits.
139
///
140
/// AbstractCircBuilder creates circuits in two phases.  First, a plan is
141
/// made for how to build the circuit.  This planning phase should be
142
/// relatively fast, and must not suspend or block.  Its purpose is to
143
/// get an early estimate of which operations the circuit will be able
144
/// to support when it's done.
145
///
146
/// Second, the circuit is actually built, using the plan as input.
147
#[async_trait]
148
pub(crate) trait AbstractCircBuilder<R: Runtime>: Send + Sync {
149
    /// The circuit type that this builder knows how to build.
150
    type Circ: AbstractCirc + Send + Sync;
151
    /// An opaque type describing how a given circuit will be built.
152
    /// It may represent some or all of a path-or it may not.
153
    //
154
    // TODO: It would be nice to have this parameterized on a lifetime,
155
    // and have that lifetime depend on the lifetime of the directory.
156
    // But I don't think that rust can do that.
157
    //
158
    // HACK(eta): I don't like the fact that `MockablePlan` is necessary here.
159
    type Plan: Send + Debug + MockablePlan;
160

            
161
    // TODO: I'd like to have a Dir type here to represent
162
    // create::DirInfo, but that would need to be parameterized too,
163
    // and would make everything complicated.
164

            
165
    /// Form a plan for how to build a new circuit that supports `usage`.
166
    ///
167
    /// Return an opaque Plan object, and a new spec describing what
168
    /// the circuit will actually support when it's built.  (For
169
    /// example, if the input spec requests a circuit that connect to
170
    /// port 80, then "planning" the circuit might involve picking an
171
    /// exit that supports port 80, and the resulting spec might be
172
    /// the exit's complete list of supported ports.)
173
    ///
174
    /// # Requirements
175
    ///
176
    /// The resulting Spec must support `usage`.
177
    fn plan_circuit(
178
        &self,
179
        usage: &TargetCircUsage,
180
        dir: DirInfo<'_>,
181
    ) -> Result<(Self::Plan, SupportedCircUsage)>;
182

            
183
    /// Construct a circuit according to a given plan.
184
    ///
185
    /// On success, return a spec describing what the circuit can be used for,
186
    /// and the circuit that was just constructed.
187
    ///
188
    /// This function should implement some kind of a timeout for
189
    /// circuits that are taking too long.
190
    ///
191
    /// # Requirements
192
    ///
193
    /// The spec that this function returns _must_ support the usage
194
    /// that was originally passed to `plan_circuit`.  It _must_ also
195
    /// contain the spec that was originally returned by
196
    /// `plan_circuit`.
197
    async fn build_circuit(
198
        &self,
199
        plan: Self::Plan,
200
    ) -> Result<(SupportedCircUsage, Arc<Self::Circ>)>;
201

            
202
    /// Return a "parallelism factor" with which circuits should be
203
    /// constructed for a given purpose.
204
    ///
205
    /// If this function returns N, then whenever we launch circuits
206
    /// for this purpose, then we launch N in parallel.
207
    ///
208
    /// The default implementation returns 1.  The value of 0 is
209
    /// treated as if it were 1.
210
604
    fn launch_parallelism(&self, usage: &TargetCircUsage) -> usize {
211
604
        let _ = usage; // default implementation ignores this.
212
604
        1
213
604
    }
214

            
215
    /// Return a "parallelism factor" for which circuits should be
216
    /// used for a given purpose.
217
    ///
218
    /// If this function returns N, then whenever we select among
219
    /// open circuits for this purpose, we choose at random from the
220
    /// best N.
221
    ///
222
    /// The default implementation returns 1.  The value of 0 is
223
    /// treated as if it were 1.
224
    // TODO: Possibly this doesn't belong in this trait.
225
382
    fn select_parallelism(&self, usage: &TargetCircUsage) -> usize {
226
382
        let _ = usage; // default implementation ignores this.
227
382
        1
228
382
    }
229

            
230
    /// Return true if we are currently attempting to learn circuit
231
    /// timeouts by building testing circuits.
232
    fn learning_timeouts(&self) -> bool;
233

            
234
    /// Flush state to the state manager if we own the lock.
235
    ///
236
    /// Return `Ok(true)` if we saved, and `Ok(false)` if we didn't hold the lock.
237
    fn save_state(&self) -> Result<bool>;
238

            
239
    /// Return this builder's [`PathConfig`](crate::PathConfig).
240
    fn path_config(&self) -> Arc<PathConfig>;
241

            
242
    /// Replace this builder's [`PathConfig`](crate::PathConfig).
243
    // TODO: This is dead_code because we only call this for the CircuitBuilder specialization of
244
    // CircMgr, not from the generic version, because this trait doesn't provide guardmgr, which is
245
    // needed by the [`CircMgr::reconfigure`] function that would be the only caller of this. We
246
    // should add `guardmgr` to this trait, make [`CircMgr::reconfigure`] generic, and remove this
247
    // dead_code marking.
248
    #[allow(dead_code)]
249
    fn set_path_config(&self, new_config: PathConfig);
250

            
251
    /// Return a reference to this builder's timeout estimator.
252
    fn estimator(&self) -> &timeouts::Estimator;
253

            
254
    /// Return a reference to this builder's `VanguardMgr`.
255
    #[cfg(feature = "vanguards")]
256
    fn vanguardmgr(&self) -> &Arc<VanguardMgr<R>>;
257

            
258
    /// Replace our state with a new owning state, assuming we have
259
    /// storage permission.
260
    fn upgrade_to_owned_state(&self) -> Result<()>;
261

            
262
    /// Reload persistent state from disk, if we don't have storage permission.
263
    fn reload_state(&self) -> Result<()>;
264

            
265
    /// Return a reference to this builder's `GuardMgr`.
266
    fn guardmgr(&self) -> &tor_guardmgr::GuardMgr<R>;
267

            
268
    /// Reconfigure this builder using the latest set of network parameters.
269
    ///
270
    /// (NOTE: for now, this only affects circuit timeout estimation.)
271
    fn update_network_parameters(&self, p: &tor_netdir::params::NetParameters);
272
}
273

            
274
/// Enumeration to track the expiration state of a circuit.
275
///
276
/// A circuit an either be unused (at which point it should expire if it is
277
/// _still unused_ by a certain time, or dirty (at which point it should
278
/// expire after a certain duration).
279
///
280
/// All circuits start out "unused" and become "dirty" when their spec
281
/// is first restricted -- that is, when they are first handed out to be
282
/// used for a request.
283
#[derive(Debug, Clone, PartialEq, Eq)]
284
enum ExpirationInfo {
285
    /// The circuit has never been used.
286
    Unused {
287
        /// A time when the circuit should expire.
288
        use_before: Instant,
289
    },
290
    /// The circuit has been used (or at least, restricted for use with a
291
    /// request) at least once.
292
    Dirty {
293
        /// The time at which this circuit's spec was first restricted.
294
        dirty_since: Instant,
295
    },
296
}
297

            
298
impl ExpirationInfo {
299
    /// Return an ExpirationInfo for a newly created circuit.
300
92
    fn new(use_before: Instant) -> Self {
301
92
        ExpirationInfo::Unused { use_before }
302
92
    }
303

            
304
    /// Mark this ExpirationInfo as dirty, if it is not already dirty.
305
448
    fn mark_dirty(&mut self, now: Instant) {
306
448
        if matches!(self, ExpirationInfo::Unused { .. }) {
307
52
            *self = ExpirationInfo::Dirty { dirty_since: now };
308
396
        }
309
448
    }
310
}
311

            
312
/// An entry for an open circuit held by an `AbstractCircMgr`.
313
#[derive(Debug, Clone)]
314
pub(crate) struct OpenEntry<C> {
315
    /// The supported usage for this circuit.
316
    spec: SupportedCircUsage,
317
    /// The circuit under management.
318
    circ: Arc<C>,
319
    /// When does this circuit expire?
320
    ///
321
    /// (Note that expired circuits are removed from the manager,
322
    /// which does not actually close them until there are no more
323
    /// references to them.)
324
    expiration: ExpirationInfo,
325
}
326

            
327
impl<C: AbstractCirc> OpenEntry<C> {
328
    /// Make a new OpenEntry for a given circuit and spec.
329
98
    fn new(spec: SupportedCircUsage, circ: Arc<C>, expiration: ExpirationInfo) -> Self {
330
98
        OpenEntry {
331
98
            spec,
332
98
            circ,
333
98
            expiration,
334
98
        }
335
98
    }
336

            
337
    /// Return true if this circuit can be used for `usage`.
338
1324
    pub(crate) fn supports(&self, usage: &TargetCircUsage) -> bool {
339
1324
        self.circ.usable() && self.spec.supports(usage)
340
1324
    }
341

            
342
    /// Change this circuit's permissible usage, based on its having
343
    /// been used for `usage` at time `now`.
344
    ///
345
    /// Return an error if this circuit may not be used for `usage`.
346
448
    fn restrict_mut(&mut self, usage: &TargetCircUsage, now: Instant) -> Result<()> {
347
448
        self.spec.restrict_mut(usage)?;
348
448
        self.expiration.mark_dirty(now);
349
448
        Ok(())
350
448
    }
351

            
352
    /// Find the "best" entry from a slice of OpenEntry for supporting
353
    /// a given `usage`.
354
    ///
355
    /// If `parallelism` is some N greater than 1, we pick randomly
356
    /// from the best `N` circuits.
357
    ///
358
    /// # Requirements
359
    ///
360
    /// Requires that `ents` is nonempty, and that every element of `ents`
361
    /// supports `spec`.
362
382
    fn find_best<'a>(
363
382
        // we do not mutate `ents`, but to return `&mut Self` we must have a mutable borrow
364
382
        ents: &'a mut [&'a mut Self],
365
382
        usage: &TargetCircUsage,
366
382
        parallelism: usize,
367
382
    ) -> &'a mut Self {
368
382
        let _ = usage; // not yet used.
369
        use rand::seq::IndexedMutRandom as _;
370
382
        let parallelism = parallelism.clamp(1, ents.len());
371
382
        // TODO: Actually look over the whole list to see which is better.
372
382
        let slice = &mut ents[0..parallelism];
373
382
        let mut rng = rand::rng();
374
382
        slice.choose_mut(&mut rng).expect("Input list was empty")
375
382
    }
376

            
377
    /// Return true if this circuit has been marked as dirty before
378
    /// `dirty_cutoff`, or if it is an unused circuit set to expire before
379
    /// `unused_cutoff`.
380
8
    fn should_expire(&self, unused_cutoff: Instant, dirty_cutoff: Instant) -> bool {
381
8
        match self.expiration {
382
            ExpirationInfo::Unused { use_before } => use_before <= unused_cutoff,
383
8
            ExpirationInfo::Dirty { dirty_since } => dirty_since <= dirty_cutoff,
384
        }
385
8
    }
386
}
387

            
388
/// A result type whose "Ok" value is the Id for a circuit from B.
389
type PendResult<B, R> = Result<<<B as AbstractCircBuilder<R>>::Circ as AbstractCirc>::Id>;
390

            
391
/// An in-progress circuit request tracked by an `AbstractCircMgr`.
392
///
393
/// (In addition to tracking circuits, `AbstractCircMgr` tracks
394
/// _requests_ for circuits.  The manager uses these entries if it
395
/// finds that some circuit created _after_ a request first launched
396
/// might meet the request's requirements.)
397
struct PendingRequest<B: AbstractCircBuilder<R>, R: Runtime> {
398
    /// Usage for the operation requested by this request
399
    usage: TargetCircUsage,
400
    /// A channel to use for telling this request about circuits that it
401
    /// might like.
402
    notify: mpsc::Sender<PendResult<B, R>>,
403
}
404

            
405
impl<B: AbstractCircBuilder<R>, R: Runtime> PendingRequest<B, R> {
406
    /// Return true if this request would be supported by `spec`.
407
34
    fn supported_by(&self, spec: &SupportedCircUsage) -> bool {
408
34
        spec.supports(&self.usage)
409
34
    }
410
}
411

            
412
/// An entry for an under-construction in-progress circuit tracked by
413
/// an `AbstractCircMgr`.
414
#[derive(Debug)]
415
struct PendingEntry<B: AbstractCircBuilder<R>, R: Runtime> {
416
    /// Specification that this circuit will support, if every pending
417
    /// request that is waiting for it is attached to it.
418
    ///
419
    /// This spec becomes more and more restricted as more pending
420
    /// requests are waiting for this circuit.
421
    ///
422
    /// This spec is contained by circ_spec, and must support the usage
423
    /// of every pending request that's waiting for this circuit.
424
    tentative_assignment: sync::Mutex<SupportedCircUsage>,
425
    /// A shared future for requests to use when waiting for
426
    /// notification of this circuit's success.
427
    receiver: Shared<oneshot::Receiver<PendResult<B, R>>>,
428
}
429

            
430
impl<B: AbstractCircBuilder<R>, R: Runtime> PendingEntry<B, R> {
431
    /// Make a new PendingEntry that starts out supporting a given
432
    /// spec.  Return that PendingEntry, along with a Sender to use to
433
    /// report the result of building this circuit.
434
140
    fn new(circ_spec: &SupportedCircUsage) -> (Self, oneshot::Sender<PendResult<B, R>>) {
435
140
        let tentative_assignment = sync::Mutex::new(circ_spec.clone());
436
140
        let (sender, receiver) = oneshot::channel();
437
140
        let receiver = receiver.shared();
438
140
        let entry = PendingEntry {
439
140
            tentative_assignment,
440
140
            receiver,
441
140
        };
442
140
        (entry, sender)
443
140
    }
444

            
445
    /// Return true if this circuit's current tentative assignment
446
    /// supports `usage`.
447
54
    fn supports(&self, usage: &TargetCircUsage) -> bool {
448
54
        let assignment = self.tentative_assignment.lock().expect("poisoned lock");
449
54
        assignment.supports(usage)
450
54
    }
451

            
452
    /// Try to change the tentative assignment of this circuit by
453
    /// restricting it for use with `usage`.
454
    ///
455
    /// Return an error if the current tentative assignment didn't
456
    /// support `usage` in the first place.
457
26
    fn tentative_restrict_mut(&self, usage: &TargetCircUsage) -> Result<()> {
458
26
        if let Ok(mut assignment) = self.tentative_assignment.lock() {
459
26
            assignment.restrict_mut(usage)?;
460
        }
461
26
        Ok(())
462
26
    }
463

            
464
    /// Find the best PendingEntry values from a slice for use with
465
    /// `usage`.
466
    ///
467
    /// # Requirements
468
    ///
469
    /// The `ents` slice must not be empty.  Every element of `ents`
470
    /// must support the given spec.
471
26
    fn find_best(ents: &[Arc<Self>], usage: &TargetCircUsage) -> Vec<Arc<Self>> {
472
26
        // TODO: Actually look over the whole list to see which is better.
473
26
        let _ = usage; // currently unused
474
26
        vec![Arc::clone(&ents[0])]
475
26
    }
476
}
477

            
478
/// Wrapper type to represent the state between planning to build a
479
/// circuit and constructing it.
480
#[derive(Debug)]
481
struct CircBuildPlan<B: AbstractCircBuilder<R>, R: Runtime> {
482
    /// The Plan object returned by [`AbstractCircBuilder::plan_circuit`].
483
    plan: B::Plan,
484
    /// A sender to notify any pending requests when this circuit is done.
485
    sender: oneshot::Sender<PendResult<B, R>>,
486
    /// A strong entry to the PendingEntry for this circuit build attempt.
487
    pending: Arc<PendingEntry<B, R>>,
488
}
489

            
490
/// The inner state of an [`AbstractCircMgr`].
491
struct CircList<B: AbstractCircBuilder<R>, R: Runtime> {
492
    /// A map from circuit ID to [`OpenEntry`] values for all managed
493
    /// open circuits.
494
    ///
495
    /// A circuit is added here from [`AbstractCircMgr::do_launch`] when we find
496
    /// that it completes successfully, and has not been cancelled.
497
    /// When we decide that such a circuit should no longer be handed out for
498
    /// any new requests, we "retire" the circuit by removing it from this map.
499
    #[allow(clippy::type_complexity)]
500
    open_circs: HashMap<<B::Circ as AbstractCirc>::Id, OpenEntry<B::Circ>>,
501
    /// Weak-set of PendingEntry for circuits that are being built.
502
    ///
503
    /// Because this set only holds weak references, and the only strong
504
    /// reference to the PendingEntry is held by the task building the circuit,
505
    /// this set's members are lazily removed after the circuit is either built
506
    /// or fails to build.
507
    ///
508
    /// This set is used for two purposes:
509
    ///
510
    /// 1. When a circuit request finds that there is no open circuit for its
511
    ///    purposes, it checks here to see if there is a pending circuit that it
512
    ///    could wait for.
513
    /// 2. When a pending circuit finishes building, it checks here to make sure
514
    ///    that it has not been cancelled. (Removing an entry from this set marks
515
    ///    it as cancelled.)
516
    ///
517
    /// An entry is added here in [`AbstractCircMgr::prepare_action`] when we
518
    /// decide that a circuit needs to be launched.
519
    ///
520
    /// Later, in [`AbstractCircMgr::do_launch`], once the circuit has finished
521
    /// (or failed), we remove the entry (by pointer identity).
522
    /// If we cannot find the entry, we conclude that the request has been
523
    /// _cancelled_, and so we discard any circuit that was created.
524
    pending_circs: PtrWeakHashSet<Weak<PendingEntry<B, R>>>,
525
    /// Weak-set of PendingRequest for requests that are waiting for a
526
    /// circuit to be built.
527
    ///
528
    /// Because this set only holds weak references, and the only
529
    /// strong reference to the PendingRequest is held by the task
530
    /// waiting for the circuit to be built, this set's members are
531
    /// lazily removed after the request succeeds or fails.
532
    pending_requests: PtrWeakHashSet<Weak<PendingRequest<B, R>>>,
533
}
534

            
535
impl<B: AbstractCircBuilder<R>, R: Runtime> CircList<B, R> {
536
    /// Make a new empty `CircList`
537
84
    fn new() -> Self {
538
84
        CircList {
539
84
            open_circs: HashMap::new(),
540
84
            pending_circs: PtrWeakHashSet::new(),
541
84
            pending_requests: PtrWeakHashSet::new(),
542
84
        }
543
84
    }
544

            
545
    /// Add `e` to the list of open circuits.
546
92
    fn add_open(&mut self, e: OpenEntry<B::Circ>) {
547
92
        let id = e.circ.id();
548
92
        self.open_circs.insert(id, e);
549
92
    }
550

            
551
    /// Find all the usable open circuits that support `usage`.
552
    ///
553
    /// Return None if there are no such circuits.
554
596
    fn find_open(&mut self, usage: &TargetCircUsage) -> Option<Vec<&mut OpenEntry<B::Circ>>> {
555
596
        let list = self.open_circs.values_mut();
556
596
        let v = SupportedCircUsage::find_supported(list, usage);
557
596
        if v.is_empty() {
558
206
            None
559
        } else {
560
390
            Some(v)
561
        }
562
596
    }
563

            
564
    /// Find an open circuit by ID.
565
    ///
566
    /// Return None if no such circuit exists in this list.
567
70
    fn get_open_mut(
568
70
        &mut self,
569
70
        id: &<B::Circ as AbstractCirc>::Id,
570
70
    ) -> Option<&mut OpenEntry<B::Circ>> {
571
70
        self.open_circs.get_mut(id)
572
70
    }
573

            
574
    /// Extract an open circuit by ID, removing it from this list.
575
    ///
576
    /// Return None if no such circuit exists in this list.
577
8
    fn take_open(&mut self, id: &<B::Circ as AbstractCirc>::Id) -> Option<OpenEntry<B::Circ>> {
578
8
        self.open_circs.remove(id)
579
8
    }
580

            
581
    /// Remove circuits based on expiration times.
582
    ///
583
    /// We remove every unused circuit that is set to expire by
584
    /// `unused_cutoff`, and every dirty circuit that has been dirty
585
    /// since before `dirty_cutoff`.
586
4
    fn expire_circs(&mut self, unused_cutoff: Instant, dirty_cutoff: Instant) {
587
4
        self.open_circs
588
8
            .retain(|_k, v| !v.should_expire(unused_cutoff, dirty_cutoff));
589
4
    }
590

            
591
    /// Remove the circuit with given `id`, if it is scheduled to
592
    /// expire now, according to the provided expiration times.
593
    fn expire_circ(
594
        &mut self,
595
        id: &<B::Circ as AbstractCirc>::Id,
596
        unused_cutoff: Instant,
597
        dirty_cutoff: Instant,
598
    ) {
599
        let should_expire = self
600
            .open_circs
601
            .get(id)
602
            .map(|v| v.should_expire(unused_cutoff, dirty_cutoff))
603
            .unwrap_or_else(|| false);
604
        if should_expire {
605
            self.open_circs.remove(id);
606
        }
607
    }
608

            
609
    /// Add `pending` to the set of in-progress circuits.
610
136
    fn add_pending_circ(&mut self, pending: Arc<PendingEntry<B, R>>) {
611
136
        self.pending_circs.insert(pending);
612
136
    }
613

            
614
    /// Find all pending circuits that support `usage`.
615
    ///
616
    /// If no such circuits are currently being built, return None.
617
166
    fn find_pending_circs(&self, usage: &TargetCircUsage) -> Option<Vec<Arc<PendingEntry<B, R>>>> {
618
166
        let result: Vec<_> = self
619
166
            .pending_circs
620
166
            .iter()
621
166
            .filter(|p| p.supports(usage))
622
166
            .filter(|p| !matches!(p.receiver.peek(), Some(Err(_))))
623
166
            .collect();
624
166

            
625
166
        if result.is_empty() {
626
140
            None
627
        } else {
628
26
            Some(result)
629
        }
630
166
    }
631

            
632
    /// Return true if `circ` is still pending.
633
    ///
634
    /// A circuit will become non-pending when finishes (successfully or not), or when it's
635
    /// removed from this list via `clear_all_circuits()`.
636
52
    fn circ_is_pending(&self, circ: &Arc<PendingEntry<B, R>>) -> bool {
637
52
        self.pending_circs.contains(circ)
638
52
    }
639

            
640
    /// Construct and add a new entry to the set of request waiting
641
    /// for a circuit.
642
    ///
643
    /// Return the request, and a new receiver stream that it should
644
    /// use for notification of possible circuits to use.
645
150
    fn add_pending_request(&mut self, pending: &Arc<PendingRequest<B, R>>) {
646
150
        self.pending_requests.insert(Arc::clone(pending));
647
150
    }
648

            
649
    /// Return all pending requests that would be satisfied by a circuit
650
    /// that supports `circ_spec`.
651
32
    fn find_pending_requests(
652
32
        &self,
653
32
        circ_spec: &SupportedCircUsage,
654
32
    ) -> Vec<Arc<PendingRequest<B, R>>> {
655
32
        self.pending_requests
656
32
            .iter()
657
34
            .filter(|pend| pend.supported_by(circ_spec))
658
32
            .collect()
659
32
    }
660

            
661
    /// Clear all pending circuits and open circuits.
662
    ///
663
    /// Calling `clear_all_circuits` ensures that any request that is answered _after
664
    /// this method runs_ will receive a circuit that was launched _after this
665
    /// method runs_.
666
    fn clear_all_circuits(&mut self) {
667
        // Note that removing entries from pending_circs will also cause the
668
        // circuit tasks to realize that they are cancelled when they
669
        // go to tell anybody about their results.
670
        self.pending_circs.clear();
671
        self.open_circs.clear();
672
    }
673
}
674

            
675
/// Timing information for circuits that have been built but never used.
676
///
677
/// Currently taken from the network parameters.
678
struct UnusedTimings {
679
    /// Minimum lifetime of a circuit created while learning
680
    /// circuit timeouts.
681
    learning: Duration,
682
    /// Minimum lifetime of a circuit created while not learning
683
    /// circuit timeouts.
684
    not_learning: Duration,
685
}
686

            
687
// This isn't really fallible, given the definitions of the underlying
688
// types.
689
#[allow(clippy::fallible_impl_from)]
690
impl From<&tor_netdir::params::NetParameters> for UnusedTimings {
691
380
    fn from(v: &tor_netdir::params::NetParameters) -> Self {
692
380
        // These try_into() calls can't fail, so unwrap() can't panic.
693
380
        #[allow(clippy::unwrap_used)]
694
380
        UnusedTimings {
695
380
            learning: v
696
380
                .unused_client_circ_timeout_while_learning_cbt
697
380
                .try_into()
698
380
                .unwrap(),
699
380
            not_learning: v.unused_client_circ_timeout.try_into().unwrap(),
700
380
        }
701
380
    }
702
}
703

            
704
/// Abstract implementation for circuit management.
705
///
706
/// The algorithm provided here is fairly simple. In its simplest form:
707
///
708
/// When somebody asks for a circuit for a given operation: if we find
709
/// one open already, we return it.  If we find in-progress circuits
710
/// that would meet our needs, we wait for one to finish (or for all
711
/// to fail).  And otherwise, we launch one or more circuits to meet the
712
/// request's needs.
713
///
714
/// If this process fails, then we retry it, up to a timeout or a
715
/// numerical limit.
716
///
717
/// If a circuit not previously considered for a given request
718
/// finishes before the request is satisfied, and if the circuit would
719
/// satisfy the request, we try to give that circuit as an answer to
720
/// that request even if it was not one of the circuits that request
721
/// was waiting for.
722
pub(crate) struct AbstractCircMgr<B: AbstractCircBuilder<R>, R: Runtime> {
723
    /// Builder used to construct circuits.
724
    builder: B,
725
    /// An asynchronous runtime to use for launching tasks and
726
    /// checking timeouts.
727
    runtime: R,
728
    /// A CircList to manage our list of circuits, requests, and
729
    /// pending circuits.
730
    circs: sync::Mutex<CircList<B, R>>,
731

            
732
    /// Configured information about when to expire circuits and requests.
733
    circuit_timing: MutCfg<CircuitTiming>,
734

            
735
    /// Minimum lifetime of an unused circuit.
736
    ///
737
    /// Derived from the network parameters.
738
    unused_timing: sync::Mutex<UnusedTimings>,
739
}
740

            
741
/// An action to take in order to satisfy a request for a circuit.
742
enum Action<B: AbstractCircBuilder<R>, R: Runtime> {
743
    /// We found an open circuit: return immediately.
744
    Open(Arc<B::Circ>),
745
    /// We found one or more pending circuits: wait until one succeeds,
746
    /// or all fail.
747
    Wait(FuturesUnordered<Shared<oneshot::Receiver<PendResult<B, R>>>>),
748
    /// We should launch circuits: here are the instructions for how
749
    /// to do so.
750
    Build(Vec<CircBuildPlan<B, R>>),
751
}
752

            
753
impl<B: AbstractCircBuilder<R> + 'static, R: Runtime> AbstractCircMgr<B, R> {
754
    /// Construct a new AbstractCircMgr.
755
76
    pub(crate) fn new(builder: B, runtime: R, circuit_timing: CircuitTiming) -> Self {
756
76
        let circs = sync::Mutex::new(CircList::new());
757
76
        let dflt_params = tor_netdir::params::NetParameters::default();
758
76
        let unused_timing = (&dflt_params).into();
759
76
        AbstractCircMgr {
760
76
            builder,
761
76
            runtime,
762
76
            circs,
763
76
            circuit_timing: circuit_timing.into(),
764
76
            unused_timing: sync::Mutex::new(unused_timing),
765
76
        }
766
76
    }
767

            
768
    /// Reconfigure this manager using the latest set of network parameters.
769
    pub(crate) fn update_network_parameters(&self, p: &tor_netdir::params::NetParameters) {
770
        let mut u = self
771
            .unused_timing
772
            .lock()
773
            .expect("Poisoned lock for unused_timing");
774
        *u = p.into();
775
    }
776

            
777
    /// Return this manager's [`CircuitTiming`].
778
604
    pub(crate) fn circuit_timing(&self) -> Arc<CircuitTiming> {
779
604
        self.circuit_timing.get()
780
604
    }
781

            
782
    /// Return this manager's [`CircuitTiming`].
783
2
    pub(crate) fn set_circuit_timing(&self, new_config: CircuitTiming) {
784
2
        self.circuit_timing.replace(new_config);
785
2
    }
786
    /// Return a circuit suitable for use with a given `usage`,
787
    /// creating that circuit if necessary, and restricting it
788
    /// under the assumption that it will be used for that spec.
789
    ///
790
    /// This is the primary entry point for AbstractCircMgr.
791
464
    pub(crate) async fn get_or_launch(
792
464
        self: &Arc<Self>,
793
464
        usage: &TargetCircUsage,
794
464
        dir: DirInfo<'_>,
795
464
    ) -> Result<(Arc<B::Circ>, CircProvenance)> {
796
        /// Largest number of "resets" that we will accept in this attempt.
797
        ///
798
        /// A "reset" is an internally generated error that does not represent a
799
        /// real problem; only a "whoops, got to try again" kind of a situation.
800
        /// For example, if we reconfigure in the middle of an attempt and need
801
        /// to re-launch the circuit, that counts as a "reset", since there was
802
        /// nothing actually _wrong_ with the circuit we were building.
803
        ///
804
        /// We accept more resets than we do real failures. However,
805
        /// we don't accept an unlimited number: we don't want to inadvertently
806
        /// permit infinite loops here. If we ever bump against this limit, we
807
        /// should not automatically increase it: we should instead figure out
808
        /// why it is happening and try to make it not happen.
809
        const MAX_RESETS: usize = 8;
810

            
811
464
        let circuit_timing = self.circuit_timing();
812
464
        let timeout_at = self.runtime.now() + circuit_timing.request_timeout;
813
464
        let max_tries = circuit_timing.request_max_retries;
814
464
        // We compute the maximum number of failures by dividing the maximum
815
464
        // number of circuits to attempt by the number that will be launched in
816
464
        // parallel for each iteration.
817
464
        let max_failures = usize::div_ceil(
818
464
            max_tries as usize,
819
464
            std::cmp::max(1, self.builder.launch_parallelism(usage)),
820
464
        );
821
464

            
822
464
        let mut retry_schedule = RetryDelay::from_msec(100);
823
464
        let mut retry_err = RetryError::<Box<Error>>::in_attempt_to("find or build a circuit");
824
464

            
825
464
        let mut n_failures = 0;
826
464
        let mut n_resets = 0;
827

            
828
540
        for attempt_num in 1.. {
829
            // How much time is remaining?
830
540
            let remaining = match timeout_at.checked_duration_since(self.runtime.now()) {
831
                None => {
832
                    retry_err.push(Error::RequestTimeout);
833
                    break;
834
                }
835
540
                Some(t) => t,
836
            };
837

            
838
540
            let error = match self.prepare_action(usage, dir, true) {
839
532
                Ok(action) => {
840
                    // We successfully found an action: Take that action.
841
532
                    let outcome = self
842
532
                        .runtime
843
532
                        .timeout(remaining, Arc::clone(self).take_action(action, usage))
844
532
                        .await;
845

            
846
528
                    match outcome {
847
448
                        Ok(Ok(circ)) => return Ok(circ),
848
80
                        Ok(Err(e)) => {
849
80
                            debug!("Circuit attempt {} failed.", attempt_num);
850
80
                            Error::RequestFailed(e)
851
                        }
852
                        Err(_) => {
853
                            // We ran out of "remaining" time; there is nothing
854
                            // more to be done.
855
4
                            warn!("All circuit attempts failed due to timeout");
856
4
                            retry_err.push(Error::RequestTimeout);
857
4
                            break;
858
                        }
859
                    }
860
                }
861
8
                Err(e) => {
862
8
                    // We couldn't pick the action!
863
8
                    debug_report!(
864
8
                        &e,
865
8
                        "Couldn't pick action for circuit attempt {}",
866
8
                        attempt_num,
867
8
                    );
868
8
                    e
869
                }
870
            };
871

            
872
            // There's been an error.  See how long we wait before we retry.
873
88
            let now = self.runtime.now();
874
88
            let retry_time =
875
88
                error.abs_retry_time(now, || retry_schedule.next_delay(&mut rand::rng()));
876

            
877
88
            let (count, count_limit) = if error.is_internal_reset() {
878
                (&mut n_resets, MAX_RESETS)
879
            } else {
880
88
                (&mut n_failures, max_failures)
881
            };
882
            // Record the error, flattening it if needed.
883
88
            match error {
884
80
                Error::RequestFailed(e) => retry_err.extend(e),
885
8
                e => retry_err.push(e),
886
            }
887

            
888
88
            *count += 1;
889
88
            // If we have reached our limit of this kind of problem, we're done.
890
88
            if *count >= count_limit {
891
4
                warn!("Reached circuit build retry limit, exiting...");
892
4
                break;
893
84
            }
894
84

            
895
84
            // Wait, or not, as appropriate.
896
84
            match retry_time {
897
76
                AbsRetryTime::Immediate => {}
898
8
                AbsRetryTime::Never => break,
899
                AbsRetryTime::At(t) => {
900
                    let remaining = timeout_at.saturating_duration_since(now);
901
                    let delay = t.saturating_duration_since(now);
902
                    self.runtime.sleep(std::cmp::min(delay, remaining)).await;
903
                }
904
            }
905
        }
906

            
907
16
        warn!("Request failed");
908
16
        Err(Error::RequestFailed(retry_err))
909
464
    }
910

            
911
    /// Make sure a circuit exists, without actually asking for it.
912
    ///
913
    /// Make sure that there is a circuit (built or in-progress) that could be
914
    /// used for `usage`, and launch one or more circuits in a background task
915
    /// if there is not.
916
    // TODO: This should probably take some kind of parallelism parameter.
917
    #[allow(dead_code)]
918
8
    pub(crate) async fn ensure_circuit(
919
8
        self: &Arc<Self>,
920
8
        usage: &TargetCircUsage,
921
8
        dir: DirInfo<'_>,
922
8
    ) -> Result<()> {
923
8
        let action = self.prepare_action(usage, dir, false)?;
924
8
        if let Action::Build(plans) = action {
925
16
            for plan in plans {
926
8
                let self_clone = Arc::clone(self);
927
8
                let _ignore_receiver = self_clone.spawn_launch(usage, plan);
928
8
            }
929
        }
930

            
931
8
        Ok(())
932
8
    }
933

            
934
    /// Choose which action we should take in order to provide a circuit
935
    /// for a given `usage`.
936
    ///
937
    /// If `restrict_circ` is true, we restrict the spec of any
938
    /// circ we decide to use to mark that it _is_ being used for
939
    /// `usage`.
940
548
    fn prepare_action(
941
548
        &self,
942
548
        usage: &TargetCircUsage,
943
548
        dir: DirInfo<'_>,
944
548
        restrict_circ: bool,
945
548
    ) -> Result<Action<B, R>> {
946
548
        let mut list = self.circs.lock().expect("poisoned lock");
947

            
948
548
        if let Some(mut open) = list.find_open(usage) {
949
            // We have open circuits that meet the spec: return the best one.
950
382
            let parallelism = self.builder.select_parallelism(usage);
951
382
            let best = OpenEntry::find_best(&mut open, usage, parallelism);
952
382
            if restrict_circ {
953
382
                let now = self.runtime.now();
954
382
                best.restrict_mut(usage, now)?;
955
            }
956
            // TODO: If we have fewer circuits here than our select
957
            // parallelism, perhaps we should launch more?
958

            
959
382
            return Ok(Action::Open(best.circ.clone()));
960
166
        }
961

            
962
166
        if let Some(pending) = list.find_pending_circs(usage) {
963
            // There are pending circuits that could meet the spec.
964
            // Restrict them under the assumption that they could all
965
            // be used for this, and then wait until one is ready (or
966
            // all have failed)
967
26
            let best = PendingEntry::find_best(&pending, usage);
968
26
            if restrict_circ {
969
52
                for item in &best {
970
                    // TODO: Do we want to tentatively restrict _all_ of these?
971
                    // not clear to me.
972
26
                    item.tentative_restrict_mut(usage)?;
973
                }
974
            }
975
26
            let stream = best.iter().map(|item| item.receiver.clone()).collect();
976
26
            // TODO: if we have fewer circuits here than our launch
977
26
            // parallelism, we might want to launch more.
978
26

            
979
26
            return Ok(Action::Wait(stream));
980
140
        }
981
140

            
982
140
        // Okay, we need to launch circuits here.
983
140
        let parallelism = std::cmp::max(1, self.builder.launch_parallelism(usage));
984
140
        let mut plans = Vec::new();
985
140
        let mut last_err = None;
986
140
        for _ in 0..parallelism {
987
140
            match self.plan_by_usage(dir, usage) {
988
132
                Ok((pending, plan)) => {
989
132
                    list.add_pending_circ(pending);
990
132
                    plans.push(plan);
991
132
                }
992
8
                Err(e) => {
993
8
                    debug!("Unable to make a plan for {:?}: {}", usage, e);
994
8
                    last_err = Some(e);
995
                }
996
            }
997
        }
998
140
        if !plans.is_empty() {
999
132
            Ok(Action::Build(plans))
8
        } else if let Some(last_err) = last_err {
8
            Err(last_err)
        } else {
            // we didn't even try to plan anything!
            Err(internal!("no plans were built, but no errors were found").into())
        }
548
    }
    /// Execute an action returned by pick-action, and return the
    /// resulting circuit or error.
532
    async fn take_action(
532
        self: Arc<Self>,
532
        act: Action<B, R>,
532
        usage: &TargetCircUsage,
532
    ) -> std::result::Result<(Arc<B::Circ>, CircProvenance), RetryError<Box<Error>>> {
        /// Store the error `err` into `retry_err`, as appropriate.
80
        fn record_error(
80
            retry_err: &mut RetryError<Box<Error>>,
80
            source: streams::Source,
80
            building: bool,
80
            mut err: Error,
80
        ) {
80
            if source == streams::Source::Right {
                // We don't care about this error, since it is from neither a circuit we launched
                // nor one that we're waiting on.
                return;
80
            }
80
            if !building {
8
                // We aren't building our own circuits, so our errors are
8
                // secondary reports of other circuits' failures.
8
                err = Error::PendingFailed(Box::new(err));
72
            }
80
            retry_err.push(err);
80
        }
        /// Return a string describing what it means, within the context of this
        /// function, to have gotten an answer from `source`.
        fn describe_source(building: bool, source: streams::Source) -> &'static str {
            match (building, source) {
                (_, streams::Source::Right) => "optimistic advice",
                (true, streams::Source::Left) => "circuit we're building",
                (false, streams::Source::Left) => "pending circuit",
            }
        }
        // Get or make a stream of futures to wait on.
532
        let (building, wait_on_stream) = match act {
382
            Action::Open(c) => {
382
                // There's already a perfectly good open circuit; we can return
382
                // it now.
382
                return Ok((c, CircProvenance::Preexisting));
            }
26
            Action::Wait(f) => {
26
                // There is one or more pending circuit that we're waiting for.
26
                // If any succeeds, we try to use it.  If they all fail, we
26
                // fail.
26
                (false, f)
            }
124
            Action::Build(plans) => {
124
                // We're going to launch one or more circuits in parallel.  We
124
                // report success if any succeeds, and failure of they all fail.
124
                let futures = FuturesUnordered::new();
248
                for plan in plans {
124
                    let self_clone = Arc::clone(&self);
124
                    // (This is where we actually launch circuits.)
124
                    futures.push(self_clone.spawn_launch(usage, plan));
124
                }
124
                (true, futures)
            }
        };
        // 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.
150
        let (pending_request, additional_stream) = {
150
            // We don't want this queue to participate in memory quota tracking.
150
            // There isn't any circuit yet, so there wouldn't be anything to account it to.
150
            // If this queue has the oldest data, probably the whole system is badly broken.
150
            // Tearing down the whole circuit manager won't help.
150
            let (send, recv) = mpsc_channel_no_memquota(8);
150
            let pending = Arc::new(PendingRequest {
150
                usage: usage.clone(),
150
                notify: send,
150
            });
150

            
150
            let mut list = self.circs.lock().expect("poisoned lock");
150
            list.add_pending_request(&pending);
150

            
150
            (pending, recv)
150
        };
150

            
150
        // We use our "select_biased" stream combiner here to ensure that:
150
        //   1) Circuits from wait_on_stream (the ones we're pending on) are
150
        //      preferred.
150
        //   2) We exit this function when those circuits are exhausted.
150
        //   3) We still get notified about other circuits that might meet our
150
        //      interests.
150
        //
150
        // The events from Left stream are the oes that we explicitly asked for,
150
        // so we'll treat errors there as real problems.  The events from the
150
        // Right stream are ones that we got opportunistically told about; it's
150
        // not a big deal if those fail.
150
        let mut incoming = streams::select_biased(wait_on_stream, additional_stream.map(Ok));
150

            
150
        let mut retry_error = RetryError::in_attempt_to("wait for circuits");
234
        while let Some((src, id)) = incoming.next().await {
150
            match id {
70
                Ok(Ok(ref id)) => {
70
                    // Great, we have a circuit. See if we can use it!
70
                    let mut list = self.circs.lock().expect("poisoned lock");
70
                    if let Some(ent) = list.get_open_mut(id) {
66
                        let now = self.runtime.now();
66
                        match ent.restrict_mut(usage, now) {
                            Ok(()) => {
                                // Great, this will work.  We drop the
                                // pending request now explicitly to remove
                                // it from the list.
66
                                drop(pending_request);
66
                                if matches!(ent.expiration, ExpirationInfo::Unused { .. }) {
                                    // Since this circuit hasn't been used yet, schedule expiration task after `max_dirtiness` from now.
                                    spawn_expiration_task(
                                        &self.runtime,
                                        Arc::downgrade(&self),
                                        ent.circ.id(),
                                        now + self.circuit_timing().max_dirtiness,
                                    );
66
                                }
66
                                return Ok((ent.circ.clone(), CircProvenance::NewlyCreated));
                            }
                            Err(e) => {
                                // In this case, a `UsageMismatched` error just means that we lost the race
                                // to restrict this circuit.
                                let e = match e {
                                    Error::UsageMismatched(e) => Error::LostUsabilityRace(e),
                                    x => x,
                                };
                                if src == streams::Source::Left {
                                    info_report!(
                                        &e,
                                        "{} suggested we use {:?}, but restrictions failed",
                                        describe_source(building, src),
                                        id,
                                    );
                                } else {
                                    debug_report!(
                                        &e,
                                        "{} suggested we use {:?}, but restrictions failed",
                                        describe_source(building, src),
                                        id,
                                    );
                                }
                                record_error(&mut retry_error, src, building, e);
                                continue;
                            }
                        }
4
                    }
                }
80
                Ok(Err(ref e)) => {
80
                    debug!("{} sent error {:?}", describe_source(building, src), e);
80
                    record_error(&mut retry_error, src, building, e.clone());
                }
                Err(oneshot::Canceled) => {
                    debug!(
                        "{} went away (Canceled), quitting take_action right away",
                        describe_source(building, src)
                    );
                    record_error(&mut retry_error, src, building, Error::PendingCanceled);
                    return Err(retry_error);
                }
            }
84
            debug!(
                "While waiting on circuit: {:?} from {}",
                id,
                describe_source(building, src)
            );
        }
        // 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.)
80
        drop(pending_request);
80

            
80
        Err(retry_error)
528
    }
    /// 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.
    #[allow(clippy::type_complexity)]
148
    fn plan_by_usage(
148
        &self,
148
        dir: DirInfo<'_>,
148
        usage: &TargetCircUsage,
148
    ) -> Result<(Arc<PendingEntry<B, R>>, CircBuildPlan<B, R>)> {
148
        let (plan, bspec) = self.builder.plan_circuit(usage, dir)?;
140
        let (pending, sender) = PendingEntry::new(&bspec);
140
        let pending = Arc::new(pending);
140

            
140
        let plan = CircBuildPlan {
140
            plan,
140
            sender,
140
            pending: Arc::clone(&pending),
140
        };
140

            
140
        Ok((pending, plan))
148
    }
    /// 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.
4
    pub(crate) fn launch_by_usage(
4
        self: &Arc<Self>,
4
        usage: &TargetCircUsage,
4
        dir: DirInfo<'_>,
4
    ) -> Result<Shared<oneshot::Receiver<PendResult<B, R>>>> {
4
        let (pending, plan) = self.plan_by_usage(dir, usage)?;
4
        self.circs
4
            .lock()
4
            .expect("Poisoned lock for circuit list")
4
            .add_pending_circ(pending);
4

            
4
        Ok(Arc::clone(self).spawn_launch(usage, plan))
4
    }
    /// 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.
136
    fn spawn_launch(
136
        self: Arc<Self>,
136
        usage: &TargetCircUsage,
136
        plan: CircBuildPlan<B, R>,
136
    ) -> Shared<oneshot::Receiver<PendResult<B, R>>> {
136
        let _ = usage; // Currently unused.
136
        let CircBuildPlan {
136
            mut plan,
136
            sender,
136
            pending,
136
        } = plan;
136
        let request_loyalty = self.circuit_timing().request_loyalty;
136

            
136
        let wait_on_future = pending.receiver.clone();
136
        let runtime = self.runtime.clone();
136
        let runtime_copy = self.runtime.clone();
136

            
136
        let tid = rand::random::<u64>();
136
        // We release this block when the circuit builder task terminates.
136
        let reason = format!("circuit builder task {}", tid);
136
        runtime.block_advance(reason.clone());
136
        // During tests, the `FakeBuilder` will need to release the block in order to fake a timeout
136
        // correctly.
136
        plan.add_blocked_advance_reason(reason);
136

            
136
        runtime
136
            .spawn(async move {
136
                let self_clone = Arc::clone(&self);
136
                let future = AssertUnwindSafe(self_clone.do_launch(plan, pending)).catch_unwind();
136
                let (new_spec, reply) = match future.await {
124
                    Ok(x) => x, // Success or regular failure
                    Err(e) => {
                        // Okay, this is a panic.  We have to tell the calling
                        // thread about it, then exit this circuit builder task.
                        let _ = sender.send(Err(internal!("circuit build task panicked").into()));
                        std::panic::panic_any(e);
                    }
                };
                // 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.)
124
                let _ = sender.send(reply.clone());
124
                if let Some(new_spec) = new_spec {
                    // 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.
52
                    let sl = runtime_copy.sleep(request_loyalty);
52
                    runtime_copy.allow_one_advance(request_loyalty);
52
                    sl.await;
32
                    let pending = {
32
                        let list = self.circs.lock().expect("poisoned lock");
32
                        list.find_pending_requests(&new_spec)
                    };
40
                    for pending_request in pending {
8
                        let _ = pending_request.notify.clone().try_send(reply.clone());
8
                    }
72
                }
104
                runtime_copy.release_advance(format!("circuit builder task {}", tid));
136
            })
136
            .expect("Couldn't spawn circuit-building task");
136

            
136
        wait_on_future
136
    }
    /// 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.
136
    async fn do_launch(
136
        self: Arc<Self>,
136
        plan: <B as AbstractCircBuilder<R>>::Plan,
136
        pending: Arc<PendingEntry<B, R>>,
136
    ) -> (Option<SupportedCircUsage>, PendResult<B, R>) {
136
        let outcome = self.builder.build_circuit(plan).await;
124
        match outcome {
72
            Err(e) => (None, Err(e)),
52
            Ok((new_spec, circ)) => {
52
                let id = circ.id();
52

            
52
                let use_duration = self.pick_use_duration();
52
                let exp_inst = self.runtime.now() + use_duration;
52
                let runtime_copy = self.runtime.clone();
52
                spawn_expiration_task(&runtime_copy, Arc::downgrade(&self), circ.id(), exp_inst);
52
                // I used to call restrict_mut here, but now I'm not so
52
                // sure. Doing restrict_mut makes sure that this
52
                // circuit will be suitable for the request that asked
52
                // for us in the first place, but that should be
52
                // ensured anyway by our tracking its tentative
52
                // assignment.
52
                //
52
                // new_spec.restrict_mut(&usage_copy).unwrap();
52
                let use_before = ExpirationInfo::new(exp_inst);
52
                let open_ent = OpenEntry::new(new_spec.clone(), circ, use_before);
52
                {
52
                    let mut list = self.circs.lock().expect("poisoned lock");
52
                    // Finally, before we return this circuit, we need to make
52
                    // sure that this pending circuit is still pending.  (If it
52
                    // is not pending, then it was cancelled through a call to
52
                    // `retire_all_circuits`, and the configuration that we used
52
                    // to launch it is now sufficiently outdated that we should
52
                    // no longer give this circuit to a client.)
52
                    if list.circ_is_pending(&pending) {
52
                        list.add_open(open_ent);
52
                        // We drop our reference to 'pending' here:
52
                        // this should make all the weak references to
52
                        // the `PendingEntry` become dangling.
52
                        drop(pending);
52
                        (Some(new_spec), Ok(id))
                    } else {
                        // This circuit is no longer pending! It must have been cancelled, probably
                        // by a call to retire_all_circuits()
                        drop(pending); // ibid
                        (None, Err(Error::CircCanceled))
                    }
                }
            }
        }
124
    }
    /// Plan and launch a new circuit to a given target, bypassing our managed
    /// pool of circuits.
    ///
    /// This method will always return a new circuit, and never return a circuit
    /// that this CircMgr gives out for anything else.
    ///
    /// The new circuit will participate in the guard and timeout apparatus as
    /// appropriate, no retry attempt will be made if the circuit fails.
    #[cfg(feature = "hs-common")]
4
    pub(crate) async fn launch_unmanaged(
4
        &self,
4
        usage: &TargetCircUsage,
4
        dir: DirInfo<'_>,
4
    ) -> Result<(SupportedCircUsage, Arc<B::Circ>)> {
4
        let (_, plan) = self.plan_by_usage(dir, usage)?;
4
        self.builder.build_circuit(plan.plan).await
4
    }
    /// 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.
8
    pub(crate) fn take_circ(&self, id: &<B::Circ as AbstractCirc>::Id) -> Option<Arc<B::Circ>> {
8
        let mut list = self.circs.lock().expect("poisoned lock");
8
        list.take_open(id).map(|e| e.circ)
8
    }
    /// Remove all open and pending circuits and from this manager, to ensure
    /// they can't be given out for any more requests.
    ///
    /// Calling `retire_all_circuits` ensures that any circuit request that gets
    /// an  answer _after this method runs_ will receive a circuit that was
    /// launched _after this method runs_.
    ///
    /// We call this method this when our configuration changes in such a way
    /// that we want to make sure that any new (or pending) requests will
    /// receive circuits that are built using the new configuration.
    //
    // For more information, see documentation on [`CircuitList::open_circs`],
    // [`CircuitList::pending_circs`], and comments in `do_launch`.
    pub(crate) fn retire_all_circuits(&self) {
        let mut list = self.circs.lock().expect("poisoned lock");
        list.clear_all_circuits();
    }
    /// 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.
4
    pub(crate) fn expire_circs(&self, now: Instant) {
4
        let mut list = self.circs.lock().expect("poisoned lock");
4
        if let Some(dirty_cutoff) = now.checked_sub(self.circuit_timing().max_dirtiness) {
4
            list.expire_circs(now, dirty_cutoff);
4
        }
4
    }
    /// Consider expiring the circuit with given circuit `id`,
    /// according to the rules in `config` and the current time `now`.
    pub(crate) fn expire_circ(&self, circ_id: &<B::Circ as AbstractCirc>::Id, now: Instant) {
        let mut list = self.circs.lock().expect("poisoned lock");
        if let Some(dirty_cutoff) = now.checked_sub(self.circuit_timing().max_dirtiness) {
            list.expire_circ(circ_id, now, dirty_cutoff);
        }
    }
    /// Return the number of open circuits held by this circuit manager.
24
    pub(crate) fn n_circs(&self) -> usize {
24
        let list = self.circs.lock().expect("poisoned lock");
24
        list.open_circs.len()
24
    }
    /// Return the number of pending circuits tracked by this circuit manager.
    #[cfg(test)]
8
    pub(crate) fn n_pending_circs(&self) -> usize {
8
        let list = self.circs.lock().expect("poisoned lock");
8
        list.pending_circs.len()
8
    }
    /// Get a reference to this manager's runtime.
16
    pub(crate) fn peek_runtime(&self) -> &R {
16
        &self.runtime
16
    }
    /// Get a reference to this manager's builder.
84
    pub(crate) fn peek_builder(&self) -> &B {
84
        &self.builder
84
    }
    /// Pick a duration by when a new circuit should expire from now
    /// if it has not yet been used
52
    fn pick_use_duration(&self) -> Duration {
52
        let timings = self
52
            .unused_timing
52
            .lock()
52
            .expect("Poisoned lock for unused_timing");
52

            
52
        if self.builder.learning_timeouts() {
            timings.learning
        } else {
            // TODO: In Tor, this calculation also depends on
            // stuff related to predicted ports and channel
            // padding.
            use tor_basic_utils::RngExt as _;
52
            let mut rng = rand::rng();
52
            rng.gen_range_checked(timings.not_learning..=timings.not_learning * 2)
52
                .expect("T .. 2x T turned out to be an empty duration range?!")
        }
52
    }
}
/// 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.
52
fn spawn_expiration_task<B, R>(
52
    runtime: &R,
52
    circmgr: Weak<AbstractCircMgr<B, R>>,
52
    circ_id: <<B as AbstractCircBuilder<R>>::Circ as AbstractCirc>::Id,
52
    exp_inst: Instant,
52
) where
52
    R: Runtime,
52
    B: 'static + AbstractCircBuilder<R>,
52
{
52
    let now = runtime.now();
52
    let rt_copy = runtime.clone();
52
    let duration = exp_inst.saturating_duration_since(now);
52

            
52
    if duration == Duration::ZERO {
        // Circuit should already expire. Expire it now.
        let cm = if let Some(cm) = Weak::upgrade(&circmgr) {
            cm
        } else {
            // Circuits manager has already been dropped, so are the references it held.
            return;
        };
        cm.expire_circ(&circ_id, now);
    } else {
        // Spawn a timer expiration task with given expiration instant.
52
        if let Err(e) = runtime.spawn(async move {
52
            rt_copy.sleep(duration).await;
            let cm = if let Some(cm) = Weak::upgrade(&circmgr) {
                cm
            } else {
                return;
            };
            cm.expire_circ(&circ_id, exp_inst);
52
        }) {
            warn_report!(e, "Unable to launch expiration task");
52
        }
    }
52
}
#[cfg(test)]
mod test {
    // @@ begin test lint list maintained by maint/add_warning @@
    #![allow(clippy::bool_assert_comparison)]
    #![allow(clippy::clone_on_copy)]
    #![allow(clippy::dbg_macro)]
    #![allow(clippy::mixed_attributes_style)]
    #![allow(clippy::print_stderr)]
    #![allow(clippy::print_stdout)]
    #![allow(clippy::single_char_pattern)]
    #![allow(clippy::unwrap_used)]
    #![allow(clippy::unchecked_duration_subtraction)]
    #![allow(clippy::useless_vec)]
    #![allow(clippy::needless_pass_by_value)]
    //! <!-- @@ end test lint list maintained by maint/add_warning @@ -->
    use super::*;
    use crate::isolation::test::{assert_isoleq, IsolationTokenEq};
    use crate::mocks::{FakeBuilder, FakeCirc, FakeId, FakeOp};
    use crate::usage::{ExitPolicy, SupportedCircUsage};
    use crate::{Error, IsolationToken, StreamIsolation, TargetCircUsage, TargetPort, TargetPorts};
    use once_cell::sync::Lazy;
    use tor_guardmgr::fallback::FallbackList;
    use tor_guardmgr::TestConfig;
    use tor_llcrypto::pk::ed25519::Ed25519Identity;
    use tor_netdir::testnet;
    use tor_persist::TestingStateMgr;
    use tor_rtcompat::SleepProvider;
    use tor_rtmock::MockRuntime;
    #[allow(deprecated)] // TODO #1885
    use tor_rtmock::MockSleepRuntime;
    static FALLBACKS_EMPTY: Lazy<FallbackList> = Lazy::new(|| [].into());
    fn di() -> DirInfo<'static> {
        (&*FALLBACKS_EMPTY).into()
    }
    fn target_to_spec(target: &TargetCircUsage) -> SupportedCircUsage {
        match target {
            TargetCircUsage::Exit {
                ports,
                isolation,
                country_code,
                require_stability,
            } => SupportedCircUsage::Exit {
                policy: ExitPolicy::from_target_ports(&TargetPorts::from(&ports[..])),
                isolation: Some(isolation.clone()),
                country_code: country_code.clone(),
                all_relays_stable: *require_stability,
            },
            _ => unimplemented!(),
        }
    }
    impl<U: PartialEq> IsolationTokenEq for OpenEntry<U> {
        fn isol_eq(&self, other: &Self) -> bool {
            self.spec.isol_eq(&other.spec)
                && self.circ == other.circ
                && self.expiration == other.expiration
        }
    }
    impl<U: PartialEq> IsolationTokenEq for &mut OpenEntry<U> {
        fn isol_eq(&self, other: &Self) -> bool {
            self.spec.isol_eq(&other.spec)
                && self.circ == other.circ
                && self.expiration == other.expiration
        }
    }
    fn make_builder<R: Runtime>(runtime: &R) -> FakeBuilder<R> {
        let state_mgr = TestingStateMgr::new();
        let guard_config = TestConfig::default();
        FakeBuilder::new(runtime, state_mgr, &guard_config)
    }
    #[test]
    fn basic_tests() {
        MockRuntime::test_with_various(|rt| async move {
            #[allow(deprecated)] // TODO #1885
            let rt = MockSleepRuntime::new(rt);
            let builder = make_builder(&rt);
            let mgr = Arc::new(AbstractCircMgr::new(
                builder,
                rt.clone(),
                CircuitTiming::default(),
            ));
            let webports = TargetCircUsage::new_from_ipv4_ports(&[80, 443]);
            // Check initialization.
            assert_eq!(mgr.n_circs(), 0);
            assert!(mgr.peek_builder().script.lock().unwrap().is_empty());
            // Launch a circuit; make sure we get it.
            let c1 = rt.wait_for(mgr.get_or_launch(&webports, di())).await;
            let c1 = c1.unwrap().0;
            assert_eq!(mgr.n_circs(), 1);
            // Make sure we get the one we already made if we ask for it.
            let port80 = TargetCircUsage::new_from_ipv4_ports(&[80]);
            let c2 = mgr.get_or_launch(&port80, di()).await;
            let c2 = c2.unwrap().0;
            assert!(FakeCirc::eq(&c1, &c2));
            assert_eq!(mgr.n_circs(), 1);
            // Now try launching two circuits "at once" to make sure that our
            // pending-circuit code works.
            let dnsport = TargetCircUsage::new_from_ipv4_ports(&[53]);
            let dnsport_restrict = TargetCircUsage::Exit {
                ports: vec![TargetPort::ipv4(53)],
                isolation: StreamIsolation::builder().build().unwrap(),
                country_code: None,
                require_stability: false,
            };
            let (c3, c4) = rt
                .wait_for(futures::future::join(
                    mgr.get_or_launch(&dnsport, di()),
                    mgr.get_or_launch(&dnsport_restrict, di()),
                ))
                .await;
            let c3 = c3.unwrap().0;
            let c4 = c4.unwrap().0;
            assert!(!FakeCirc::eq(&c1, &c3));
            assert!(FakeCirc::eq(&c3, &c4));
            assert_eq!(c3.id(), c4.id());
            assert_eq!(mgr.n_circs(), 2);
            // Now we're going to remove c3 from consideration.  It's the
            // same as c4, so removing c4 will give us None.
            let c3_taken = mgr.take_circ(&c3.id()).unwrap();
            let now_its_gone = mgr.take_circ(&c4.id());
            assert!(FakeCirc::eq(&c3_taken, &c3));
            assert!(now_its_gone.is_none());
            assert_eq!(mgr.n_circs(), 1);
            // Having removed them, let's launch another dnsport and make
            // sure we get a different circuit.
            let c5 = rt.wait_for(mgr.get_or_launch(&dnsport, di())).await;
            let c5 = c5.unwrap().0;
            assert!(!FakeCirc::eq(&c3, &c5));
            assert!(!FakeCirc::eq(&c4, &c5));
            assert_eq!(mgr.n_circs(), 2);
            // Now try launch_by_usage.
            let prev = mgr.n_pending_circs();
            assert!(mgr.launch_by_usage(&dnsport, di()).is_ok());
            assert_eq!(mgr.n_pending_circs(), prev + 1);
            // TODO: Actually make sure that launch_by_usage launched
            // the right thing.
        });
    }
    #[test]
    fn request_timeout() {
        MockRuntime::test_with_various(|rt| async move {
            #[allow(deprecated)] // TODO #1885
            let rt = MockSleepRuntime::new(rt);
            let ports = TargetCircUsage::new_from_ipv4_ports(&[80, 443]);
            // This will fail once, and then completely time out.  The
            // result will be a failure.
            let builder = make_builder(&rt);
            builder.set(&ports, vec![FakeOp::Fail, FakeOp::Timeout]);
            let mgr = Arc::new(AbstractCircMgr::new(
                builder,
                rt.clone(),
                CircuitTiming::default(),
            ));
            let c1 = mgr
                .peek_runtime()
                .wait_for(mgr.get_or_launch(&ports, di()))
                .await;
            assert!(matches!(c1, Err(Error::RequestFailed(_))));
        });
    }
    #[test]
    fn request_timeout2() {
        MockRuntime::test_with_various(|rt| async move {
            #[allow(deprecated)] // TODO #1885
            let rt = MockSleepRuntime::new(rt);
            // Now try a more complicated case: we'll try to get things so
            // that we wait for a little over our predicted time because
            // of our wait-for-next-action logic.
            let ports = TargetCircUsage::new_from_ipv4_ports(&[80, 443]);
            let builder = make_builder(&rt);
            builder.set(
                &ports,
                vec![
                    FakeOp::Delay(Duration::from_millis(60_000 - 25)),
                    FakeOp::NoPlan,
                ],
            );
            let mgr = Arc::new(AbstractCircMgr::new(
                builder,
                rt.clone(),
                CircuitTiming::default(),
            ));
            let c1 = mgr
                .peek_runtime()
                .wait_for(mgr.get_or_launch(&ports, di()))
                .await;
            assert!(matches!(c1, Err(Error::RequestFailed(_))));
        });
    }
    #[test]
    fn request_unplannable() {
        MockRuntime::test_with_various(|rt| async move {
            #[allow(deprecated)] // TODO #1885
            let rt = MockSleepRuntime::new(rt);
            let ports = TargetCircUsage::new_from_ipv4_ports(&[80, 443]);
            // This will fail a the planning stages, a lot.
            let builder = make_builder(&rt);
            builder.set(&ports, vec![FakeOp::NoPlan; 2000]);
            let mgr = Arc::new(AbstractCircMgr::new(
                builder,
                rt.clone(),
                CircuitTiming::default(),
            ));
            let c1 = rt.wait_for(mgr.get_or_launch(&ports, di())).await;
            assert!(matches!(c1, Err(Error::RequestFailed(_))));
        });
    }
    #[test]
    fn request_fails_too_much() {
        MockRuntime::test_with_various(|rt| async move {
            #[allow(deprecated)] // TODO #1885
            let rt = MockSleepRuntime::new(rt);
            let ports = TargetCircUsage::new_from_ipv4_ports(&[80, 443]);
            // This will fail 1000 times, which is above the retry limit.
            let builder = make_builder(&rt);
            builder.set(&ports, vec![FakeOp::Fail; 1000]);
            let mgr = Arc::new(AbstractCircMgr::new(
                builder,
                rt.clone(),
                CircuitTiming::default(),
            ));
            let c1 = rt.wait_for(mgr.get_or_launch(&ports, di())).await;
            assert!(matches!(c1, Err(Error::RequestFailed(_))));
        });
    }
    #[test]
    fn request_wrong_spec() {
        MockRuntime::test_with_various(|rt| async move {
            #[allow(deprecated)] // TODO #1885
            let rt = MockSleepRuntime::new(rt);
            let ports = TargetCircUsage::new_from_ipv4_ports(&[80, 443]);
            // The first time this is called, it will build a circuit
            // with the wrong spec.  (A circuit builder should never
            // actually _do_ that, but it's something we code for.)
            let builder = make_builder(&rt);
            builder.set(
                &ports,
                vec![FakeOp::WrongSpec(target_to_spec(
                    &TargetCircUsage::new_from_ipv4_ports(&[22]),
                ))],
            );
            let mgr = Arc::new(AbstractCircMgr::new(
                builder,
                rt.clone(),
                CircuitTiming::default(),
            ));
            let c1 = rt.wait_for(mgr.get_or_launch(&ports, di())).await;
            assert!(c1.is_ok());
        });
    }
    #[test]
    fn request_retried() {
        MockRuntime::test_with_various(|rt| async move {
            #[allow(deprecated)] // TODO #1885
            let rt = MockSleepRuntime::new(rt);
            let ports = TargetCircUsage::new_from_ipv4_ports(&[80, 443]);
            // This will fail twice, and then succeed. The result will be
            // a success.
            let builder = make_builder(&rt);
            builder.set(&ports, vec![FakeOp::Fail, FakeOp::Fail]);
            let mgr = Arc::new(AbstractCircMgr::new(
                builder,
                rt.clone(),
                CircuitTiming::default(),
            ));
            // This test doesn't exercise any timeout behaviour.
            rt.block_advance("test doesn't require advancing");
            let (c1, c2) = rt
                .wait_for(futures::future::join(
                    mgr.get_or_launch(&ports, di()),
                    mgr.get_or_launch(&ports, di()),
                ))
                .await;
            let c1 = c1.unwrap().0;
            let c2 = c2.unwrap().0;
            assert!(FakeCirc::eq(&c1, &c2));
        });
    }
    #[test]
    fn isolated() {
        MockRuntime::test_with_various(|rt| async move {
            #[allow(deprecated)] // TODO #1885
            let rt = MockSleepRuntime::new(rt);
            let builder = make_builder(&rt);
            let mgr = Arc::new(AbstractCircMgr::new(
                builder,
                rt.clone(),
                CircuitTiming::default(),
            ));
            // Set our isolation so that iso1 and iso2 can't share a circuit,
            // but no_iso can share a circuit with either.
            let iso1 = TargetCircUsage::Exit {
                ports: vec![TargetPort::ipv4(443)],
                isolation: StreamIsolation::builder()
                    .owner_token(IsolationToken::new())
                    .build()
                    .unwrap(),
                country_code: None,
                require_stability: false,
            };
            let iso2 = TargetCircUsage::Exit {
                ports: vec![TargetPort::ipv4(443)],
                isolation: StreamIsolation::builder()
                    .owner_token(IsolationToken::new())
                    .build()
                    .unwrap(),
                country_code: None,
                require_stability: false,
            };
            let no_iso1 = TargetCircUsage::new_from_ipv4_ports(&[443]);
            let no_iso2 = no_iso1.clone();
            // We're going to try launching these circuits in 24 different
            // orders, to make sure that the outcome is correct each time.
            use itertools::Itertools;
            let timeouts: Vec<_> = [0_u64, 2, 4, 6]
                .iter()
                .map(|d| Duration::from_millis(*d))
                .collect();
            for delays in timeouts.iter().permutations(4) {
                let d1 = delays[0];
                let d2 = delays[1];
                let d3 = delays[2];
                let d4 = delays[2];
                let (c_iso1, c_iso2, c_no_iso1, c_no_iso2) = rt
                    .wait_for(futures::future::join4(
                        async {
                            rt.sleep(*d1).await;
                            mgr.get_or_launch(&iso1, di()).await
                        },
                        async {
                            rt.sleep(*d2).await;
                            mgr.get_or_launch(&iso2, di()).await
                        },
                        async {
                            rt.sleep(*d3).await;
                            mgr.get_or_launch(&no_iso1, di()).await
                        },
                        async {
                            rt.sleep(*d4).await;
                            mgr.get_or_launch(&no_iso2, di()).await
                        },
                    ))
                    .await;
                let c_iso1 = c_iso1.unwrap().0;
                let c_iso2 = c_iso2.unwrap().0;
                let c_no_iso1 = c_no_iso1.unwrap().0;
                let c_no_iso2 = c_no_iso2.unwrap().0;
                assert!(!FakeCirc::eq(&c_iso1, &c_iso2));
                assert!(!FakeCirc::eq(&c_iso1, &c_no_iso1));
                assert!(!FakeCirc::eq(&c_iso1, &c_no_iso2));
                assert!(!FakeCirc::eq(&c_iso2, &c_no_iso1));
                assert!(!FakeCirc::eq(&c_iso2, &c_no_iso2));
                assert!(FakeCirc::eq(&c_no_iso1, &c_no_iso2));
            }
        });
    }
    #[test]
    fn opportunistic() {
        MockRuntime::test_with_various(|rt| async move {
            #[allow(deprecated)] // TODO #1885
            let rt = MockSleepRuntime::new(rt);
            // The first request will time out completely, but we're
            // making a second request after we launch it.  That
            // request should succeed, and notify the first request.
            let ports1 = TargetCircUsage::new_from_ipv4_ports(&[80]);
            let ports2 = TargetCircUsage::new_from_ipv4_ports(&[80, 443]);
            let builder = make_builder(&rt);
            builder.set(&ports1, vec![FakeOp::Timeout]);
            let mgr = Arc::new(AbstractCircMgr::new(
                builder,
                rt.clone(),
                CircuitTiming::default(),
            ));
            // Note that ports2 will be wider than ports1, so the second
            // request will have to launch a new circuit.
            let (c1, c2) = rt
                .wait_for(futures::future::join(
                    mgr.get_or_launch(&ports1, di()),
                    async {
                        rt.sleep(Duration::from_millis(100)).await;
                        mgr.get_or_launch(&ports2, di()).await
                    },
                ))
                .await;
            if let (Ok((c1, _)), Ok((c2, _))) = (c1, c2) {
                assert!(FakeCirc::eq(&c1, &c2));
            } else {
                panic!();
            };
        });
    }
    #[test]
    fn prebuild() {
        MockRuntime::test_with_various(|rt| async move {
            // This time we're going to use ensure_circuit() to make
            // sure that a circuit gets built, and then launch two
            // other circuits that will use it.
            #[allow(deprecated)] // TODO #1885
            let rt = MockSleepRuntime::new(rt);
            let builder = make_builder(&rt);
            let mgr = Arc::new(AbstractCircMgr::new(
                builder,
                rt.clone(),
                CircuitTiming::default(),
            ));
            let ports1 = TargetCircUsage::new_from_ipv4_ports(&[80, 443]);
            let ports2 = TargetCircUsage::new_from_ipv4_ports(&[80]);
            let ports3 = TargetCircUsage::new_from_ipv4_ports(&[443]);
            let (ok, c1, c2) = rt
                .wait_for(futures::future::join3(
                    mgr.ensure_circuit(&ports1, di()),
                    async {
                        rt.sleep(Duration::from_millis(10)).await;
                        mgr.get_or_launch(&ports2, di()).await
                    },
                    async {
                        rt.sleep(Duration::from_millis(50)).await;
                        mgr.get_or_launch(&ports3, di()).await
                    },
                ))
                .await;
            assert!(ok.is_ok());
            let c1 = c1.unwrap().0;
            let c2 = c2.unwrap().0;
            // If we had launched these separately, they wouldn't share
            // a circuit.
            assert!(FakeCirc::eq(&c1, &c2));
        });
    }
    #[test]
    fn expiration() {
        MockRuntime::test_with_various(|rt| async move {
            use crate::config::CircuitTimingBuilder;
            // Now let's make some circuits -- one dirty, one clean, and
            // make sure that one expires and one doesn't.
            #[allow(deprecated)] // TODO #1885
            let rt = MockSleepRuntime::new(rt);
            let builder = make_builder(&rt);
            let circuit_timing = CircuitTimingBuilder::default()
                .max_dirtiness(Duration::from_secs(15))
                .build()
                .unwrap();
            let mgr = Arc::new(AbstractCircMgr::new(builder, rt.clone(), circuit_timing));
            let imap = TargetCircUsage::new_from_ipv4_ports(&[993]);
            let pop = TargetCircUsage::new_from_ipv4_ports(&[995]);
            let (ok, pop1) = rt
                .wait_for(futures::future::join(
                    mgr.ensure_circuit(&imap, di()),
                    mgr.get_or_launch(&pop, di()),
                ))
                .await;
            assert!(ok.is_ok());
            let pop1 = pop1.unwrap().0;
            rt.advance(Duration::from_secs(30)).await;
            rt.advance(Duration::from_secs(15)).await;
            let imap1 = rt.wait_for(mgr.get_or_launch(&imap, di())).await.unwrap().0;
            // This should expire the pop circuit, since it came from
            // get_or_launch() [which marks the circuit as being
            // used].  It should not expire the imap circuit, since
            // it was not dirty until 15 seconds after the cutoff.
            let now = rt.now();
            mgr.expire_circs(now);
            let (pop2, imap2) = rt
                .wait_for(futures::future::join(
                    mgr.get_or_launch(&pop, di()),
                    mgr.get_or_launch(&imap, di()),
                ))
                .await;
            let pop2 = pop2.unwrap().0;
            let imap2 = imap2.unwrap().0;
            assert!(!FakeCirc::eq(&pop2, &pop1));
            assert!(FakeCirc::eq(&imap2, &imap1));
        });
    }
    /// Returns three exit policies; one that permits nothing, one that permits ports 80
    /// and 443 only, and one that permits all ports.
    fn get_exit_policies() -> (ExitPolicy, ExitPolicy, ExitPolicy) {
        // FIXME(eta): the below is copypasta; would be nice to have a better way of
        //             constructing ExitPolicy objects for testing maybe
        let network = testnet::construct_netdir().unwrap_if_sufficient().unwrap();
        // Nodes with ID 0x0a through 0x13 and 0x1e through 0x27 are
        // exits.  Odd-numbered ones allow only ports 80 and 443;
        // even-numbered ones allow all ports.
        let id_noexit: Ed25519Identity = [0x05; 32].into();
        let id_webexit: Ed25519Identity = [0x11; 32].into();
        let id_fullexit: Ed25519Identity = [0x20; 32].into();
        let not_exit = network.by_id(&id_noexit).unwrap();
        let web_exit = network.by_id(&id_webexit).unwrap();
        let full_exit = network.by_id(&id_fullexit).unwrap();
        let ep_none = ExitPolicy::from_relay(&not_exit);
        let ep_web = ExitPolicy::from_relay(&web_exit);
        let ep_full = ExitPolicy::from_relay(&full_exit);
        (ep_none, ep_web, ep_full)
    }
    #[test]
    fn test_find_supported() {
        let (ep_none, ep_web, ep_full) = get_exit_policies();
        let fake_circ = Arc::new(FakeCirc { id: FakeId::next() });
        let expiration = ExpirationInfo::Unused {
            use_before: Instant::now() + Duration::from_secs(60 * 60),
        };
        let mut entry_none = OpenEntry::new(
            SupportedCircUsage::Exit {
                policy: ep_none,
                isolation: None,
                country_code: None,
                all_relays_stable: true,
            },
            fake_circ.clone(),
            expiration.clone(),
        );
        let mut entry_none_c = entry_none.clone();
        let mut entry_web = OpenEntry::new(
            SupportedCircUsage::Exit {
                policy: ep_web,
                isolation: None,
                country_code: None,
                all_relays_stable: true,
            },
            fake_circ.clone(),
            expiration.clone(),
        );
        let mut entry_web_c = entry_web.clone();
        let mut entry_full = OpenEntry::new(
            SupportedCircUsage::Exit {
                policy: ep_full,
                isolation: None,
                country_code: None,
                all_relays_stable: true,
            },
            fake_circ,
            expiration,
        );
        let mut entry_full_c = entry_full.clone();
        let usage_web = TargetCircUsage::new_from_ipv4_ports(&[80]);
        let empty: Vec<&mut OpenEntry<FakeCirc>> = vec![];
        assert_isoleq!(
            SupportedCircUsage::find_supported(vec![&mut entry_none].into_iter(), &usage_web),
            empty
        );
        // HACK(eta): We have to faff around with clones and such because
        //            `abstract_spec_find_supported` has a silly signature that involves `&mut`
        //            refs, which we can't have more than one of.
        assert_isoleq!(
            SupportedCircUsage::find_supported(
                vec![&mut entry_none, &mut entry_web].into_iter(),
                &usage_web,
            ),
            vec![&mut entry_web_c]
        );
        assert_isoleq!(
            SupportedCircUsage::find_supported(
                vec![&mut entry_none, &mut entry_web, &mut entry_full].into_iter(),
                &usage_web,
            ),
            vec![&mut entry_web_c, &mut entry_full_c]
        );
        // Test preemptive circuit usage:
        let usage_preemptive_web = TargetCircUsage::Preemptive {
            port: Some(TargetPort::ipv4(80)),
            circs: 2,
            require_stability: false,
        };
        let usage_preemptive_dns = TargetCircUsage::Preemptive {
            port: None,
            circs: 2,
            require_stability: false,
        };
        // shouldn't return anything unless there are >=2 circuits
        assert_isoleq!(
            SupportedCircUsage::find_supported(
                vec![&mut entry_none].into_iter(),
                &usage_preemptive_web
            ),
            empty
        );
        assert_isoleq!(
            SupportedCircUsage::find_supported(
                vec![&mut entry_none].into_iter(),
                &usage_preemptive_dns
            ),
            empty
        );
        assert_isoleq!(
            SupportedCircUsage::find_supported(
                vec![&mut entry_none, &mut entry_web].into_iter(),
                &usage_preemptive_web
            ),
            empty
        );
        assert_isoleq!(
            SupportedCircUsage::find_supported(
                vec![&mut entry_none, &mut entry_web].into_iter(),
                &usage_preemptive_dns
            ),
            vec![&mut entry_none_c, &mut entry_web_c]
        );
        assert_isoleq!(
            SupportedCircUsage::find_supported(
                vec![&mut entry_none, &mut entry_web, &mut entry_full].into_iter(),
                &usage_preemptive_web
            ),
            vec![&mut entry_web_c, &mut entry_full_c]
        );
    }
    #[test]
    fn test_circlist_preemptive_target_circs() {
        MockRuntime::test_with_various(|rt| async move {
            #[allow(deprecated)] // TODO #1885
            let rt = MockSleepRuntime::new(rt);
            let netdir = testnet::construct_netdir().unwrap_if_sufficient().unwrap();
            let dirinfo = DirInfo::Directory(&netdir);
            let builder = make_builder(&rt);
            for circs in [2, 8].iter() {
                let mut circlist = CircList::<FakeBuilder<MockRuntime>, MockRuntime>::new();
                let preemptive_target = TargetCircUsage::Preemptive {
                    port: Some(TargetPort::ipv4(80)),
                    circs: *circs,
                    require_stability: false,
                };
                for _ in 0..*circs {
                    assert!(circlist.find_open(&preemptive_target).is_none());
                    let usage = TargetCircUsage::new_from_ipv4_ports(&[80]);
                    let (plan, _) = builder.plan_circuit(&usage, dirinfo).unwrap();
                    let (spec, circ) = rt.wait_for(builder.build_circuit(plan)).await.unwrap();
                    let entry = OpenEntry::new(
                        spec,
                        circ,
                        ExpirationInfo::new(rt.now() + Duration::from_secs(60)),
                    );
                    circlist.add_open(entry);
                }
                assert!(circlist.find_open(&preemptive_target).is_some());
            }
        });
    }
}