//! Code to handle incoming cells on a circuit.

//!

//! ## On message validation

//!

//! There are three steps for validating an incoming message on a stream:

//!

//! 1. Is the message contextually appropriate? (e.g., no more than one

//!    `CONNECTED` message per stream.) This is handled by calling

//!    [`CmdChecker::check_msg`](crate::stream::CmdChecker::check_msg).

//! 2. Does the message comply with flow-control rules? (e.g., no more data than

//!    we've gotten SENDMEs for.) For open streams, the stream itself handles

//!    this; for half-closed streams, the reactor handles it using the

//!    `halfstream` module.

//! 3. Does the message have an acceptable command type, and is the message

//!    well-formed? For open streams, the streams themselves handle this check.

//!    For half-closed streams, the reactor handles it by calling

//!    `consume_checked_msg()`.

pub(super) mod circuit;

mod conflux;

mod control;

pub(super) mod syncview;

use crate::crypto::cell::HopNum;

use crate::crypto::handshake::ntor_v3::NtorV3PublicKey;

use crate::memquota::{CircuitAccount, StreamAccount};

use crate::stream::AnyCmdChecker;

#[cfg(feature = "hs-service")]

use crate::stream::{IncomingStreamRequest, IncomingStreamRequestFilter};

use crate::tunnel::circuit::celltypes::ClientCircChanMsg;

use crate::tunnel::circuit::unique_id::UniqId;

use crate::tunnel::circuit::CircuitRxReceiver;

use crate::tunnel::{streammap, HopLocation, TargetHop};

use crate::util::err::ReactorError;

use crate::util::skew::ClockSkew;

use crate::{Error, Result};

use circuit::{Circuit, CircuitCmd};

use conflux::ConfluxSet;

use control::ControlHandler;

use std::cmp::Ordering;

use std::collections::BinaryHeap;

use std::mem::size_of;

use tor_cell::relaycell::msg::{AnyRelayMsg, End, Sendme};

use tor_cell::relaycell::{AnyRelayMsgOuter, RelayCellFormat, StreamId, UnparsedRelayMsg};

use tor_error::{bad_api_usage, internal, into_bad_api_usage, Bug};

use tor_rtcompat::DynTimeProvider;

use futures::channel::mpsc;

use futures::StreamExt;

use futures::{select_biased, FutureExt as _};

use oneshot_fused_workaround as oneshot;

use std::result::Result as StdResult;

use std::sync::Arc;

use crate::channel::Channel;

use crate::crypto::handshake::ntor::{NtorClient, NtorPublicKey};

use crate::tunnel::circuit::{StreamMpscReceiver, StreamMpscSender};

use derive_deftly::Deftly;

use derive_more::From;

use tor_cell::chancell::CircId;

use tor_llcrypto::pk;

use tor_memquota::mq_queue::{self, MpscSpec};

use tor_memquota::{derive_deftly_template_HasMemoryCost, memory_cost_structural_copy};

use tracing::trace;

use super::circuit::{MutableState, TunnelMutableState};

#[cfg(feature = "conflux")]

use {crate::util::err::ConfluxHandshakeError, conflux::OooRelayMsg};

pub(super) use control::CtrlCmd;

pub(super) use control::CtrlMsg;

/// The type of a oneshot channel used to inform reactor of the result of an operation.

pub(super) type ReactorResultChannel<T> = oneshot::Sender<Result<T>>;

/// Contains a list of conflux handshake results.

#[cfg(feature = "conflux")]

pub(super) type ConfluxHandshakeResult = Vec<StdResult<(), ConfluxHandshakeError>>;

/// The type of oneshot channel used to inform reactor users of the outcome

/// of a client-side conflux handshake.

///

/// Contains a list of handshake results, one for each circuit that we were asked

/// to link in the tunnel.

#[cfg(feature = "conflux")]

pub(super) type ConfluxLinkResultChannel = ReactorResultChannel<ConfluxHandshakeResult>;

pub(crate) use circuit::{RECV_WINDOW_INIT, STREAM_READER_BUFFER};

/// MPSC queue containing stream requests

#[cfg(feature = "hs-service")]

type StreamReqSender = mq_queue::Sender<StreamReqInfo, MpscSpec>;

/// A handshake type, to be used when creating circuit hops.

#[derive(Clone, Debug)]

pub(crate) enum CircuitHandshake {

    /// Use the CREATE_FAST handshake.

    CreateFast,

    /// Use the ntor handshake.

    Ntor {

        /// The public key of the relay.

        public_key: NtorPublicKey,

        /// The Ed25519 identity of the relay, which is verified against the

        /// identity held in the circuit's channel.

        ed_identity: pk::ed25519::Ed25519Identity,

    },

    /// Use the ntor-v3 handshake.

    NtorV3 {

        /// The public key of the relay.

        public_key: NtorV3PublicKey,

    },

}

/// A behavior to perform when closing a stream.

///

/// We don't use `Option<End>`` here, since the behavior of `SendNothing` is so surprising

/// that we shouldn't let it pass unremarked.

#[derive(Clone, Debug)]

pub(crate) enum CloseStreamBehavior {

    /// Send nothing at all, so that the other side will not realize we have

    /// closed the stream.

    ///

    /// We should only do this for incoming onion service streams when we

    /// want to black-hole the client's requests.

    SendNothing,

    /// Send an End cell, if we haven't already sent one.

    SendEnd(End),

}

impl Default for CloseStreamBehavior {

}

// TODO(conflux): the RunOnceCmd/RunOnceCmdInner/CircuitCmd/CircuitAction enum

// proliferation is a bit bothersome, but unavoidable with the current design.

//

// We should consider getting rid of some of these enums (if possible),

// and coming up with more intuitive names.

/// One or more [`RunOnceCmdInner`] to run inside [`Reactor::run_once`].

#[derive(From, Debug)]

enum RunOnceCmd {

    /// Run a single `RunOnceCmdInner` command.

    Single(RunOnceCmdInner),

    /// Run multiple `RunOnceCmdInner` commands.

    //

    // Note: this whole enum *could* be replaced with Vec<RunOnceCmdInner>,

    // but most of the time we're only going to have *one* RunOnceCmdInner

    // to run per run_once() loop. The enum enables us avoid the extra heap

    // allocation for the `RunOnceCmd::Single` case.

    Multiple(Vec<RunOnceCmdInner>),

}

/// Instructions for running something in the reactor loop.

///

/// Run at the end of [`Reactor::run_once`].

//

// TODO: many of the variants of this enum have an identical CtrlMsg counterpart.

// We should consider making each variant a tuple variant and deduplicating the fields.

#[derive(educe::Educe)]

#[educe(Debug)]

enum RunOnceCmdInner {

    /// Send a RELAY cell.

    Send {

        /// The leg the cell should be sent on.

        leg: LegId,

        /// The cell to send.

        cell: SendRelayCell,

        /// A channel for sending completion notifications.

        done: Option<ReactorResultChannel<()>>,

    },

    /// Send a given control message on this circuit, and install a control-message handler to

    /// receive responses.

    #[cfg(feature = "send-control-msg")]

    SendMsgAndInstallHandler {

        /// The message to send, if any

        msg: Option<AnyRelayMsgOuter>,

        /// A message handler to install.

        ///

        /// If this is `None`, there must already be a message handler installed

        #[educe(Debug(ignore))]

        handler: Option<Box<dyn MetaCellHandler + Send + 'static>>,

        /// A sender that we use to tell the caller that the message was sent

        /// and the handler installed.

        done: oneshot::Sender<Result<()>>,

    },

    /// Handle a SENDME message.

    HandleSendMe {

        /// The leg the SENDME was received on.

        leg: LegId,

        /// The hop number.

        hop: HopNum,

        /// The SENDME message to handle.

        sendme: Sendme,

    },

    /// Begin a stream with the provided hop in this circuit.

    ///

    /// Uses the provided stream ID, and sends the provided message to that hop.

    BeginStream {

        /// The cell to send.

        cell: Result<(SendRelayCell, StreamId)>,

        /// The location of the hop on the tunnel. We don't use this (and `Circuit`s shouldn't need

        /// to worry about legs anyways), but need it so that we can pass it back in `done` to the

        /// caller.

        hop: HopLocation,

        /// The circuit leg to begin the stream on.

        leg: LegId,

        /// Oneshot channel to notify on completion, with the allocated stream ID.

        done: ReactorResultChannel<(StreamId, HopLocation, RelayCellFormat)>,

    },

    /// Close the specified stream.

    CloseStream {

        /// The hop number.

        hop: HopLocation,

        /// The ID of the stream to close.

        sid: StreamId,

        /// The stream-closing behavior.

        behav: CloseStreamBehavior,

        /// The reason for closing the stream.

        reason: streammap::TerminateReason,

        /// A channel for sending completion notifications.

        done: Option<ReactorResultChannel<()>>,

    },

    /// Get the clock skew claimed by the first hop of the circuit.

    FirstHopClockSkew {

        /// Oneshot channel to return the clock skew.

        answer: oneshot::Sender<StdResult<ClockSkew, Bug>>,

    },

    /// Remove a circuit leg from the conflux set.

    RemoveLeg {

        /// The circuit leg to remove.

        leg: LegId,

        /// The reason for removal.

        ///

        /// This is only used for conflux circuits that get removed

        /// before the conflux handshake is complete.

        ///

        /// The [`RemoveLegReason`] is mapped by the reactor to a

        /// [`ConfluxHandshakeError`] that is sent to the initiator of the

        /// handshake to indicate the reason the handshake failed.

        reason: RemoveLegReason,

    },

    /// A circuit has completed the conflux handshake,

    /// and wants to send the specified cell.

    ///

    /// This is similar to [`RunOnceCmdInner::Send`],

    /// but needs to remain a separate variant,

    /// because in addition to instructing the reactor to send a cell,

    /// it also notifies it that the conflux handshake is complete on the specified `leg`.

    /// This enables the reactor to save the handshake result (`Ok(())`),

    /// and, if there are no other legs still in the handshake phase,

    /// send the result to the handshake initiator.

    #[cfg(feature = "conflux")]

    ConfluxHandshakeComplete {

        /// The circuit leg that has completed the handshake,

        /// This is the leg the cell should be sent on.

        leg: LegId,

        /// The cell to send.

        cell: SendRelayCell,

    },

    /// Send a LINK cell on each of the unlinked circuit legs in the conflux set of this reactor.

    #[cfg(feature = "conflux")]

    Link {

        /// The circuits to link into the tunnel

        #[educe(Debug(ignore))]

        circuits: Vec<Circuit>,

        /// Oneshot channel for notifying of conflux handshake completion.

        answer: ConfluxLinkResultChannel,

    },

    /// Enqueue an out-of-order cell in ooo_msg.

    #[cfg(feature = "conflux")]

    Enqueue {

        /// The leg the entry originated from.

        leg: LegId,

        /// The out-of-order message.

        msg: OooRelayMsg,

    },

    /// Perform a clean shutdown on this circuit.

    CleanShutdown,

}

impl RunOnceCmdInner {

    /// Create a [`RunOnceCmdInner`] out of a [`CircuitCmd`] and [`LegIdKey`].

            CircuitCmd::CloseStream {

            #[cfg(feature = "conflux")]

            #[cfg(feature = "conflux")]

            #[cfg(feature = "conflux")]

        }

}

/// Cmd for sending a relay cell.

///

/// The contents of this struct are passed to `send_relay_cell`

#[educe(Debug)]

pub(crate) struct SendRelayCell {

    /// The hop number.

    pub(crate) hop: HopNum,

    /// Whether to use a RELAY_EARLY cell.

    pub(crate) early: bool,

    /// The cell to send.

    pub(crate) cell: AnyRelayMsgOuter,

}

/// A command to execute at the end of [`Reactor::run_once`].

#[derive(From, Debug)]

#[allow(clippy::large_enum_variant)] // TODO #2003: should we resolve this?

enum CircuitAction {

    /// Run a single `CircuitCmd` command.

    RunCmd {

        /// The unique identifier of the circuit leg to run the command on

        leg: LegIdKey,

        /// The command to run.

        cmd: CircuitCmd,

    },

    /// Handle a control message

    HandleControl(CtrlMsg),

    /// Handle an input message.

    HandleCell {

        /// The unique identifier of the circuit leg the message was received on.

        leg: LegIdKey,

        /// The message to handle.

        cell: ClientCircChanMsg,

    },

    /// Remove the specified circuit leg from the conflux set.

    ///

    /// Returned whenever a single circuit leg needs to be be removed

    /// from the reactor's conflux set, without necessarily tearing down

    /// the whole set or shutting down the reactor.

    ///

    /// Note: this action *can* cause the reactor to shut down

    /// (and the conflux set to be closed).

    ///

    /// See the [`ConfluxSet::remove`] docs for more on the exact behavior of this command.

    RemoveLeg {

        /// The leg to remove.

        leg: LegIdKey,

        /// The reason for removal.

        ///

        /// This is only used for conflux circuits that get removed

        /// before the conflux handshake is complete.

        ///

        /// The [`RemoveLegReason`] is mapped by the reactor to a

        /// [`ConfluxHandshakeError`] that is sent to the initiator of the

        /// handshake to indicate the reason the handshake failed.

        reason: RemoveLegReason,

    },

}

/// The reason for removing a circuit leg from the conflux set.

#[derive(Debug)]

enum RemoveLegReason {

    /// The conflux handshake timed out.

    ///

    /// On the client-side, this means we didn't receive

    /// the CONFLUX_LINKED response in time.

    ConfluxHandshakeTimeout,

    /// An error occurred during conflux handshake.

    ConfluxHandshakeErr(Error),

    /// The channel was closed.

    ChannelClosed,

}

/// An object that's waiting for a meta cell (one not associated with a stream) in order to make

/// progress.

///

/// # Background

///

/// The `Reactor` can't have async functions that send and receive cells, because its job is to

/// send and receive cells: if one of its functions tried to do that, it would just hang forever.

///

/// To get around this problem, the reactor can send some cells, and then make one of these

/// `MetaCellHandler` objects, which will be run when the reply arrives.

pub(crate) trait MetaCellHandler: Send {

    /// The hop we're expecting the message to come from. This is compared against the hop

    /// from which we actually receive messages, and an error is thrown if the two don't match.

    fn expected_hop(&self) -> HopNum;

    /// Called when the message we were waiting for arrives.

    ///

    /// Gets a copy of the `Reactor` in order to do anything it likes there.

    ///

    /// If this function returns an error, the reactor will shut down.

    fn handle_msg(

        &mut self,

        msg: UnparsedRelayMsg,

        reactor: &mut Circuit,

    ) -> Result<MetaCellDisposition>;

}

/// A possible successful outcome of giving a message to a [`MsgHandler`](super::msghandler::MsgHandler).

#[derive(Debug, Clone)]

#[cfg_attr(feature = "send-control-msg", visibility::make(pub))]

#[non_exhaustive]

pub(crate) enum MetaCellDisposition {

    /// The message was consumed; the handler should remain installed.

    #[cfg(feature = "send-control-msg")]

    Consumed,

    /// The message was consumed; the handler should be uninstalled.

    ConversationFinished,

    /// The message was consumed; the circuit should be closed.

    #[cfg(feature = "send-control-msg")]

    CloseCirc,

    // TODO: Eventually we might want the ability to have multiple handlers

    // installed, and to let them say "not for me, maybe for somebody else?".

    // But right now we don't need that.

}

/// A unique identifier for a circuit leg.

///

/// After the circuit is torn down, its `LegId` becomes invalid.

/// The same `LegId` won't be reused for a future circuit.

//

// TODO(#1857): make this pub

#[allow(unused)]

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

#[derive_deftly(HasMemoryCost)]

pub(crate) struct LegId(pub(crate) LegIdKey);

// TODO(#1999): can we use `UniqId` as the key instead of this newtype?

//

// See https://gitlab.torproject.org/tpo/core/arti/-/merge_requests/2996#note_3199069

slotmap_careful::new_key_type! {

    /// A key type for the circuit leg slotmap

    ///

    /// See [`LegId`].

    pub(crate) struct LegIdKey;

}

impl From<LegIdKey> for LegId {

}

memory_cost_structural_copy!(LegIdKey);

/// Unwrap the specified [`Option`], returning a [`ReactorError::Shutdown`] if it is `None`.

///

/// This is a macro instead of a function to work around borrowck errors

/// in the select! from run_once().

macro_rules! unwrap_or_shutdown {

    ($self:expr, $res:expr, $reason:expr) => {{

        match $res {

            None => {

                trace!("{}: reactor shutdown due to {}", $self.unique_id, $reason);

                Err(ReactorError::Shutdown)

            }

            Some(v) => Ok(v),

        }

    }};

}

/// Object to handle incoming cells and background tasks on a circuit

///

/// This type is returned when you finish a circuit; you need to spawn a

/// new task that calls `run()` on it.

#[must_use = "If you don't call run() on a reactor, the circuit won't work."]

pub struct Reactor {

    /// Receiver for control messages for this reactor, sent by `ClientCirc` objects.

    ///

    /// This channel is polled in [`Reactor::run_once`], but only if the `chan_sender` sink

    /// is ready to accept cells.

    control: mpsc::UnboundedReceiver<CtrlMsg>,

    /// Receiver for command messages for this reactor, sent by `ClientCirc` objects.

    ///

    /// This channel is polled in [`Reactor::run_once`].

    ///

    /// NOTE: this is a separate channel from `control`, because some messages

    /// have higher priority and need to be handled even if the `chan_sender` is not

    /// ready (whereas `control` messages are not read until the `chan_sender` sink

    /// is ready to accept cells).

    command: mpsc::UnboundedReceiver<CtrlCmd>,

    /// A oneshot sender that is used to alert other tasks when this reactor is

    /// finally dropped.

    ///

    /// It is a sender for Void because we never actually want to send anything here;

    /// we only want to generate canceled events.

    #[allow(dead_code)] // the only purpose of this field is to be dropped.

    reactor_closed_tx: oneshot::Sender<void::Void>,

    /// A set of circuits that form a tunnel.

    ///

    /// Contains 1 or more circuits.

    ///

    /// Circuits may be added to this set throughout the lifetime of the reactor.

    ///

    /// Sometimes, the reactor will remove circuits from this set,

    /// for example if the `LINKED` message takes too long to arrive,

    /// or if congestion control negotiation fails.

    /// The reactor will continue running with the remaining circuits.

    /// It will shut down if *all* the circuits are removed.

    ///

    // TODO(conflux): document all the reasons why the reactor might

    // chose to tear down a circuit or tunnel (timeouts, protocol violations, etc.)

    circuits: ConfluxSet,

    /// An identifier for logging about this reactor's circuit.

    unique_id: UniqId,

    /// Handlers, shared with `Circuit`.

    cell_handlers: CellHandlers,

    /// The time provider, used for conflux handshake timeouts.

    runtime: DynTimeProvider,

    /// The conflux handshake context, if there is an on-going handshake.

    ///

    /// Set to `None` if this is a single-path tunnel,

    /// or if none of the circuit legs from our conflux set

    /// are currently in the conflux handshake phase.

    #[cfg(feature = "conflux")]

    conflux_hs_ctx: Option<ConfluxHandshakeCtx>,

    /// A min-heap buffering all the out-of-order messages received so far.

    ///

    /// TODO(conflux): this becomes a DoS vector unless we impose a limit

    /// on its size. We should make this participate in the memquota memory

    /// tracking system, somehow.

    #[cfg(feature = "conflux")]

    ooo_msgs: BinaryHeap<ConfluxHeapEntry>,

}

/// The context for an on-going conflux handshake.

#[cfg(feature = "conflux")]

struct ConfluxHandshakeCtx {

    /// A channel for notifying the caller of the outcome of a CONFLUX_LINK request.

    answer: ConfluxLinkResultChannel,

    /// The number of legs that are currently doing the handshake.

    num_legs: usize,

    /// The handshake results we have collected so far.

    results: ConfluxHandshakeResult,

}

/// An out-of-order message buffered in [`Reactor::ooo_msgs`].

#[derive(Debug)]

#[cfg(feature = "conflux")]

struct ConfluxHeapEntry {

    /// The leg id this message came from.

    leg_id: LegId,

    /// The out of order message

    msg: OooRelayMsg,

}

#[cfg(feature = "conflux")]

impl Ord for ConfluxHeapEntry {

}

#[cfg(feature = "conflux")]

impl PartialOrd for ConfluxHeapEntry {

}

#[cfg(feature = "conflux")]

impl PartialEq for ConfluxHeapEntry {

}

#[cfg(feature = "conflux")]

impl Eq for ConfluxHeapEntry {}

/// Cell handlers, shared between the Reactor and its underlying `Circuit`s.

struct CellHandlers {

    /// A handler for a meta cell, together with a result channel to notify on completion.

    meta_handler: Option<Box<dyn MetaCellHandler + Send>>,

    /// A handler for incoming stream requests.

    #[cfg(feature = "hs-service")]

    incoming_stream_req_handler: Option<IncomingStreamRequestHandler>,

}

/// Information about an incoming stream request.

#[cfg(feature = "hs-service")]

#[derive(Debug, Deftly)]

#[derive_deftly(HasMemoryCost)]

pub(crate) struct StreamReqInfo {

    /// The [`IncomingStreamRequest`].

    pub(crate) req: IncomingStreamRequest,

    /// The ID of the stream being requested.

    pub(crate) stream_id: StreamId,

    /// The [`HopNum`].

    //

    // TODO: When we add support for exit relays, we need to turn this into an Option<HopNum>.

    // (For outbound messages (towards relays), there is only one hop that can send them: the client.)

    //

    // TODO: For onion services, we might be able to enforce the HopNum earlier: we would never accept an

    // incoming stream request from two separate hops.  (There is only one that's valid.)

    pub(crate) hop_num: HopNum,

    /// The [`LegId`] of the circuit the request came on.

    pub(crate) leg: LegId,

    /// The format which must be used with this stream to encode messages.

    #[deftly(has_memory_cost(indirect_size = "0"))]

    pub(crate) relay_cell_format: RelayCellFormat,

    /// A channel for receiving messages from this stream.

    #[deftly(has_memory_cost(indirect_size = "0"))] // estimate

    pub(crate) receiver: StreamMpscReceiver<UnparsedRelayMsg>,

    /// A channel for sending messages to be sent on this stream.

    #[deftly(has_memory_cost(indirect_size = "size_of::<AnyRelayMsg>()"))] // estimate

    pub(crate) msg_tx: StreamMpscSender<AnyRelayMsg>,

    /// The memory quota account to be used for this stream

    #[deftly(has_memory_cost(indirect_size = "0"))] // estimate (it contains an Arc)

    pub(crate) memquota: StreamAccount,

}

/// Data required for handling an incoming stream request.

#[cfg(feature = "hs-service")]

#[educe(Debug)]

struct IncomingStreamRequestHandler {

    /// A sender for sharing information about an incoming stream request.

    incoming_sender: StreamReqSender,

    /// A [`AnyCmdChecker`] for validating incoming stream requests.

    cmd_checker: AnyCmdChecker,

    /// The hop to expect incoming stream requests from.

    hop_num: HopNum,

    /// An [`IncomingStreamRequestFilter`] for checking whether the user wants

    /// this request, or wants to reject it immediately.

    #[educe(Debug(ignore))]

    filter: Box<dyn IncomingStreamRequestFilter>,

}

impl Reactor {

    /// Create a new circuit reactor.

    ///

    /// The reactor will send outbound messages on `channel`, receive incoming

    /// messages on `input`, and identify this circuit by the channel-local

    /// [`CircId`] provided.

    ///

    /// The internal unique identifier for this circuit will be `unique_id`.

    #[allow(clippy::type_complexity)] // TODO

    /// Launch the reactor, and run until the circuit closes or we

    /// encounter an error.

    ///

    /// Once this method returns, the circuit is dead and cannot be

    /// used again.

            }

        };

    /// Helper for run: doesn't mark the circuit closed on finish.  Only

    /// processes one cell or control message.

                self.unique_id

            );

            },

            // Check whether we've got a control message pending.

            //

            // Note: unfortunately, reading from control here means we might start

            // handling control messages before our chan_senders are ready.

            // With the current design, this is inevitable: we can't know which circuit leg

            // a control message is meant for without first reading the control message from

            // the channel, and at that point, we can't know for sure whether that particular

            // circuit is ready for sending.

            },

        };

                } else {

                    // TODO(conflux): we return RunOnceCmd::Multiple even if there's a single command.

                    //

                    // See the TODO(conflux) on `Circuit::handle_cell`.

                }

            }

        };

        // Check if it's time to switch our primary leg.

        //

        // TODO(conflux): this only be done just before sending a cell.

        //

        // See https://gitlab.torproject.org/tpo/core/arti/-/merge_requests/2946#note_3192013

        #[cfg(feature = "conflux")]

    /// Try to process the previously-out-of-order messages we might have buffered.

    #[cfg(feature = "conflux")]

        // Check if we're ready to dequeue any of the previously out-of-order cells.

        }

    /// Handle a [`RunOnceCmd`].

                // While we know `sendable` is ready to accept *one* cell,

                // we can't be certain it will be able to accept *all* of the cells

                // that need to be sent here. This means we *may* end up buffering

                // in its underlying SometimesUnboundedSink! That is OK, because

                // RunOnceCmd::Multiple is only used for handling packed cells.

                }

            }

        }

    /// Handle a [`RunOnceCmd`].

                // TODO: check the cc window

            }

            #[cfg(feature = "send-control-msg")]

                        // TODO(conflux): let the RunOnceCmdInner specify which leg to send the cell on

                        // don't care if receiver goes away.

                    }

                    }

                }

            }

            RunOnceCmdInner::BeginStream {

                        // don't care if receiver goes away.

                    }

                    }

                }

            }

            RunOnceCmdInner::CloseStream {

                    }

                };

            }

                // NOTE: it's okay to await. We are only awaiting on the congestion_signals

                // future which *should* resolve immediately

            }

            RunOnceCmdInner::CleanShutdown => {

            }

                    };

            }

            #[cfg(feature = "conflux")]

            }

            #[cfg(feature = "conflux")]

            }

            #[cfg(feature = "conflux")]

        }

    /// Wait for a [`CtrlMsg::Create`] to come along to set up the circuit.

    ///

    /// Returns an error if an unexpected `CtrlMsg` is received.

                    #[cfg(test)]

                    CtrlCmd::AddFakeHop {

                    } => {

                    },

                    _ => {

                    }

                }

            },

        };

            CtrlMsg::Create {

            } => {

                // TODO(conflux): instead of crashing the reactor, it might be better

                // to send the error via the done channel instead

            }

            _ => {

            }

        }

    /// Add the specified handshake result to our `ConfluxHandshakeContext`.

    ///

    /// If all the circuits we were waiting on have finished the conflux handshake,

    /// the `ConfluxHandshakeContext` is consumed, and the results we have collected

    /// are sent to the handshake initiator.

    #[cfg(feature = "conflux")]

            }

            None => {

            }

        };

            // Time to remove the conflux handshake context

            // and extract the results we have collected

            // We don't expect to receive any more handshake results,

            // at least not until we get another LinkCircuits control message,

            // which will install a new ConfluxHandshakeCtx with a channel

            // for us to send updates on

    /// Prepare a `SendRelayCell` request, and install the given meta-cell handler.

    /// Handle a shutdown request.

            self.unique_id

        );

    /// Handle a request to shutdown the reactor and return the only [`Circuit`] in this tunnel.

    ///

    /// Returns an error over the `answer` channel if the reactor has no circuits,

    /// or more than one circuit. The reactor will shut down regardless.

    #[cfg(feature = "conflux")]

    /// Resolves a [`TargetHop`] to a [`HopLocation`].

    ///

    /// After resolving a `TargetHop::LastHop`,

    /// the `HopLocation` can become stale if a single-path circuit is later extended or truncated.

    /// This means that the `HopLocation` can become stale from one reactor iteration to the next.

    ///

    /// It's generally okay to hold on to a (possibly stale) `HopLocation`

    /// if you need a fixed hop position in the tunnel.

    /// For example if we open a stream to `TargetHop::LastHop`,

    /// we would want to store the stream position as a `HopLocation` and not a `TargetHop::LastHop`

    /// as we don't want the stream position to change as the tunnel is extended or truncated.

    ///

    /// Returns [`NoHopsBuiltError`] if trying to resolve `TargetHop::LastHop`

    /// and the tunnel has no hops

    /// (either has no legs, or has legs which contain no hops).

            TargetHop::LastHop => {

                    // single-path tunnel

                    // multi-path tunnel

                } else {

                    // no legs

                }

            }

        }

    /// Resolves a [`HopLocation`] to a [`LegId`] and [`HopNum`].

    ///

    /// After resolving a `HopLocation::JoinPoint`,

    /// the [`LegId`] and [`HopNum`] can become stale if the primary leg changes.

    ///

    /// You should try to only resolve to a specific [`LegId`] and [`HopNum`] immediately before you

    /// need them,

    /// and you should not hold on to the resolved [`LegId`] and [`HopNum`] between reactor

    /// iterations as the primary leg may change from one iteration to the next.

    ///

    /// Returns [`NoJoinPointError`] if trying to resolve `HopLocation::JoinPoint`

    /// but it does not have a join point.

            HopLocation::JoinPoint => {

                } else {

                    // Attempted to get the join point of a non-multipath tunnel.

                }

            }

        }

    /// Does congestion control use stream SENDMEs for the given hop?

    ///

    /// Returns `None` if either the `leg` or `hop` don't exist.

    /// Handle a request to link some extra circuits in the reactor's conflux set.

    ///

    /// The circuits are validated, and if they do not have the same length,

    /// or if they do not all have the same last hop, an error is returned on

    /// the `answer` channel, and the conflux handshake is *not* initiated.

    ///

    /// If validation succeeds, the circuits are added to this reactor's conflux set,

    /// and the conflux handshake is initiated (by sending a LINK cell on each leg).

    ///

    /// NOTE: this blocks the reactor main loop until all the cells are sent.

    #[cfg(feature = "conflux")]