//! A `maybenot`-specific backend for padding.

// Some of the circuit padding implementation isn't reachable unless

// the extra-experimental circ-padding-manual feature is also present.

//

// TODO circpad: Remove this once we have circ-padding negotiation implemented.

#![cfg_attr(

    all(feature = "circ-padding", not(feature = "circ-padding-manual")),

    expect(dead_code)

)]

mod backend;

use std::collections::VecDeque;

use std::num::NonZeroU16;

use std::pin::Pin;

use std::sync::{Arc, Mutex};

use std::task::{Context, Poll, Waker};

use bitvec::BitArr;

use maybenot::MachineId;

use smallvec::SmallVec;

use tor_memquota::memory_cost_structural_copy;

use tor_rtcompat::{DynTimeProvider, SleepProvider};

use crate::HopNum;

use crate::circuit::HOPS;

use crate::util::err::ExcessPadding;

use backend::PaddingBackend;

/// The type of Instant that we'll use for our padding machines.

///

/// We use a separate type alias here in case we want to move to coarsetime.

type Instant = std::time::Instant;

/// The type of Duration that we'll use for our padding machines.

///

/// We use a separate type alias here in case we want to move to coarsetime.

type Duration = std::time::Duration;

/// A type we use to generate a set of [`PaddingEvent`].

///

/// This is a separate type so we can tune it and make it into a smallvec if needed.

type PaddingEventQueue = VecDeque<PaddingEvent>;

/// A type we use to generate a set of [`PaddingEvent`].

///

/// This is a separate type so we can tune it and make it into a smallvec if needed.

type PerHopPaddingEventVec = Vec<PerHopPaddingEvent>;

/// Specifications for a set of maybenot padding machines as used in Arti: used to construct a `maybenot::Framework`.

#[derive(Clone, Debug, derive_builder::Builder)]

#[builder(build_fn(

    validate = "Self::validate",

    private,

    error = "CircuitPadderConfigError"

))]

#[builder(name = "CircuitPadderConfig")]

#[cfg_attr(not(feature = "circ-padding-manual"), builder(private))]

#[cfg_attr(feature = "circ-padding-manual", builder(public))]

pub(crate) struct PaddingRules {

    /// List of padding machines to use for shaping traffic.

    ///

    /// Note that this list may be empty, if we only want to receive padding,

    /// and never send it.

    machines: Arc<[maybenot::Machine]>,

    /// Maximum allowable outbound padding fraction.

    ///

    /// Passed directly to maybenot; not enforced in Arti.

    /// See [`maybenot::Framework::new`] for details.

    ///

    /// Must be between 0.0 and 1.0

    #[builder(default = "1.0")]

    max_outbound_blocking_frac: f64,

    /// Maximum allowable outbound blocking fraction.

    ///

    /// Passed directly to maybenot; not enforced in Arti.

    /// See [`maybenot::Framework::new`] for details.

    ///

    /// Must be between 0.0 and 1.0.

    #[builder(default = "1.0")]

    max_outbound_padding_frac: f64,

    /// Maximum allowable fraction of inbound padding

    #[builder(default = "1.0")]

    max_inbound_padding_frac: f64,

    /// Number of cells before which we should not enforce max_inbound_padding_frac.

    #[builder(default = "1")]

    enforce_inbound_padding_after_cells: u16,

}

/// An error returned from validating a [`CircuitPadderConfig`].

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

#[cfg_attr(feature = "circ-padding-manual", visibility::make(pub))]

#[non_exhaustive]

pub(crate) enum CircuitPadderConfigError {

    /// A field needed to be given, but wasn't.

    #[error("No value was given for {0}")]

    UninitializedField(&'static str),

    /// A field needed to be a proper fraction, but wasn't.

    #[error("Value was out of range for {0}. (Must be between 0 and 1)")]

    FractionOutOfRange(&'static str),

    /// Maybenot gave us an error when initializing the framework.

    #[error("Maybenot could not initialize framework for rules")]

    MaybenotError(#[from] maybenot::Error),

}

impl From<derive_builder::UninitializedFieldError> for CircuitPadderConfigError {

}

impl CircuitPadderConfig {

    /// Helper: Return an error if this is not a valid Builder.

        macro_rules! enforce_frac {

            { $field:ident } =>

            {

                    return Err(CircuitPadderConfigError::FractionOutOfRange(stringify!($field)));

                }

            }

        }

    /// Construct a [`CircuitPadder`] based on this [`CircuitPadderConfig`].

    ///

    /// A [`CircuitPadderConfig`] is created its accessors, and used with this method to build a [`CircuitPadder`].

    ///

    /// That [`CircuitPadder`] can then be installed on a circuit using [`ClientCirc::start_padding_at_hop`](crate::client::circuit::ClientCirc::start_padding_at_hop).

}

impl PaddingRules {

    /// Create a [`PaddingBackend`] for this [`PaddingRules`], so we can install it in a

    /// [`PaddingShared`].

        // TODO circpad: specialize this for particular values of n_machines,

        // when we finally go to implement padding.

        const OPTIMIZE_FOR_N_MACHINES: usize = 4;

    /// Create a new `PaddingStats` to reflect the rules for inbound padding of this  PaddingRules

}

/// A opaque handle to a padding implementation for a single hop.

///

/// This type is constructed with [`CircuitPadderConfig::create_padder`].

#[derive(derive_more::Debug)]

#[cfg_attr(feature = "circ-padding-manual", visibility::make(pub))]

pub(crate) struct CircuitPadder {

    /// The initial padding stats and restrictions for inbound padding.

    initial_stats: PaddingStats,

    /// The underlying backend to use.

    #[debug(skip)]

    backend: Box<dyn PaddingBackend>,

}

/// An instruction from the padding machine to the circuit.

///

/// These are returned from the [`PaddingEventStream`].

///

/// When the `circ-padding` feature is disabled, these won't actually be constructed.

#[derive(Clone, Copy, Debug)]

pub(crate) enum PaddingEvent {

    /// An instruction to send padding.

    SendPadding(SendPadding),

    /// An instruction to start blocking outbound traffic,

    /// or change the hop at which traffic is blocked.

    StartBlocking(StartBlocking),

    /// An instruction to stop all blocking.

    StopBlocking,

}

/// An instruction from a single padding hop.

///

/// This will be turned into a [`PaddingEvent`] before it's given

/// to the circuit reactor.

#[derive(Clone, Copy, Debug)]

enum PerHopPaddingEvent {

    /// An instruction to send padding.

    SendPadding {

        /// The machine that told us to send the padding.

        ///

        /// (We need to use this when we report that we sent the padding.)

        machine: MachineId,

        /// Whether the padding can be replaced with regular data.

        replace: Replace,

        /// Whether the padding can bypass a bypassable block.

        bypass: Bypass,

    },

    /// An instruction to start blocking traffic..

    StartBlocking {

        /// Whether this blocking instance may be bypassed by padding with

        /// [`Bypass::BypassBlocking`].

        ///

        /// (Note that this is _not_ a `Bypass`, since that enum notes whether

        /// or not a _padding_ cell can bypass blocking)

        is_bypassable: bool,

    },

    /// An instruction to stop blocking.

    StopBlocking,

}

/// Whether a given piece of padding can be replaced with queued data.

///

/// This is an enum to avoid confusing it with `Bypass`.

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

pub(crate) enum Replace {

    /// The padding can be replaced

    /// either by packaging data in a regular data cell,

    /// or with data currently queued but not yet sent.

    Replaceable,

    /// The padding must be queued; it can't be replaced with data.

    NotReplaceable,

}

impl Replace {

    /// Construct a [`Replace`] from a bool.

        }

}

/// Whether a piece of padding can bypass a bypassable case of blocking.

///

/// This is an enum to avoid confusing it with `Release`.

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

pub(crate) enum Bypass {

    /// This padding may bypass the block, if the block is bypassable.

    ///

    /// Note that this case has complicated interactions with `Replace`; see the

    /// `maybenot` documentation.

    BypassBlocking,

    /// The padding may not bypass the block.

    DoNotBypass,

}

/// Information about a queued cell that we need to feed back into the padding

/// subsystem.

#[derive(Clone, Copy, Debug)]

pub(crate) struct QueuedCellPaddingInfo {

    /// The hop that will receive this cell.

    pub(crate) target_hop: HopNum,

}

memory_cost_structural_copy!(QueuedCellPaddingInfo);

impl Bypass {

    /// Construct a [`Bypass`] from a bool.

        }

}

/// An indication that we should send a padding cell.

///

/// Don't drop this: instead, once the cell is queued,

/// pass this `SendPadding` object to the relevant [`PaddingController`]

/// to report that the particular piece of padding has been queued.

#[derive(Clone, Debug, Copy)]

pub(crate) struct SendPadding {

    /// The machine within a framework that told us to send the padding.

    ///

    /// We store this so we can tell the framework which machine's padding we sent.

    machine: maybenot::MachineId,

    /// The hop to which we need to send the padding.

    pub(crate) hop: HopNum,

    /// Whether this padding can be replaced by regular data.

    pub(crate) replace: Replace,

    /// Whether this padding cell should bypass any current blocking.

    pub(crate) bypass: Bypass,

}

impl SendPadding {

    /// Convert this SendPadding into a TriggerEvent for Maybenot,

    /// to indicate that the padding was sent.

    /// If true, we are allowed to replace this padding cell

    /// with a normal non-padding cell.

    ///

    /// (If we do, we should call [`PaddingController::queued_data_as_padding`])

    /// Return whether this padding cell is allowed to bypass any current blocking.

}

/// An instruction to start blocking traffic

/// or to change the rules for blocking traffic.

#[derive(Clone, Copy, Debug)]

pub(crate) struct StartBlocking {

    /// If true, then padding traffic _to the blocking hop_

    /// can bypass this block, if it has [`Bypass::BypassBlocking`].

    ///

    /// (All traffic can be sent to earlier hops as normal.

    /// No traffic may be sent to later hops.)

    pub(crate) is_bypassable: bool,

}

/// Absolute upper bound for number of hops.

const MAX_HOPS: usize = 64;

/// A handle to the padding state of a single circuit.

///

/// Used to tell the padders about events that they need to react to.

#[derive(Clone, derive_more::Debug)]

pub(crate) struct PaddingController<S = DynTimeProvider>

where

    S: SleepProvider,

{

    /// The underlying shared state.

    #[debug(skip)]

    shared: Arc<Mutex<PaddingShared<S>>>,

}

/// The shared state for a single circuit's padding.

///

/// Used by both PaddingController and PaddingEventStream.

struct PaddingShared<S: SleepProvider> {

    /// A sleep provider for telling the time and creating sleep futures.

    runtime: S,

    /// Per-hop state for each hop that we have enabled padding with.

    ///

    /// INVARIANT: the length of this vector is no greater than `MAX_HOPS`.

    hops: SmallVec<[Option<Box<dyn PaddingBackend>>; HOPS]>,

    /// Records about how much padding and normal traffic we have received from each hop,

    /// and how much padding is allowed.

    stats: SmallVec<[Option<PaddingStats>; HOPS]>,

    /// Which hops are currently blocking, and whether that blocking is bypassable.

    blocking: BlockingState,

    /// When will the currently pending sleep future next expire?

    ///

    /// We keep track of this so that we know when we need to reset the sleep future.

    /// It gets updated by `PaddingStream::schedule_next_wakeup`,

    /// which we call in `<PaddingStream as Stream>::poll_next` immediately

    /// before we create a timer.

    next_scheduled_wakeup: Option<Instant>,

    /// A deque of `PaddingEvent` that we want to yield from our [`PaddingEventStream`].

    ///

    /// NOTE: If you put new items in this list from anywhere other than inside

    /// `PaddingEventStream::poll_next`, you need to alert the `waker`.

    pending_events: PaddingEventQueue,

    /// A waker to alert if we've added any events to padding_events,

    /// or if we need the stream to re-poll.

    //

    // TODO circpad: This waker is redundant with the one stored in every backend's `Timer`.

    // When we revisit this code we may want to consider combining them somehow.

    waker: Waker,

}

/// The number of padding and non-padding cells we have received from each hop,

/// and the rules for how many are allowed.

#[derive(Clone, Debug)]

struct PaddingStats {

    /// The number of padding cells we've received from this hop.

    n_padding: u64,

    /// The number of non-padding cells we've received from this hop.

    n_normal: u64,

    /// The maximum allowable fraction of padding cells.

    max_padding_frac: f32,

    /// A lower limit, below which we will not enforce `max_padding_frac`.

    //

    // This is a NonZero for several reasons:

    // - It doesn't make sense to enforce a ratio when no cells have been received.

    // - If we only check when the total is at above zero, we can avoid a division-by-zero check.

    // - Having an impossible value here ensures that the niche optimization

    //   will work on PaddingStats.

    enforce_max_after: NonZeroU16,

}

impl PaddingStats {

    /// Return an error if this PaddingStats has exceeded its maximum.

        // Total number of cells.

        // (It is impossible to get so many cells that this addition will overflow a u64.)

            // TODO: is there a way we can avoid a floating-point op here?

            // Or can we limit the number of times that we need to check?

            // (Tobias suggests randomization; I'm worried.)

            //

            // On the one hand, this may never appear on our profiles.

            // But on the other hand, if it _does_ matter for performance,

            // it is likely to be on some marginal platform with bad FP performance,

            // where we are unlikely to be doing much testing.

            //

            // One promising possibility is to calculate a minimum amount of padding

            // that we _know_ will be valid, given the current total,

            // and then not check again until we at all until we reach that amount.

}

/// Current padding-related blocking status for a circuit.

///

/// We have to keep track of whether each hop is blocked or not,

/// and whether its blocking is bypassable.

/// But all we actually need to tell the reactor code

/// is whether to block the _entire_ circuit or not.

//

// TODO circpad: It might beneficial

// to block only the first blocking hop and its successors,

// but that creates tricky starvation problems

// in the case where we have queued traffic for a later, blocking, hop

// that prevents us from flushing any messages to earlier hops.

// We could solve this with tricky out-of-order designs,

// but for now we're just treating "blocked" as a boolean.

#[derive(Default)]

struct BlockingState {

    /// Whether each hop is currently blocked.

    hop_blocked: BitArr![for MAX_HOPS],

    /// Whether each hop's blocking is currently **not** bypassable.

    blocking_non_bypassable: BitArr![for MAX_HOPS],

}

impl BlockingState {

    /// Set the hop at position `idx` to blocked.

    /// Set the hop at position `idx` to unblocked.

    /// Return a [`PaddingEvent`]

            // At least one hop has non-bypassable blocking, so our blocking is non-bypassable.

            // At least one hop is blocking, but no hop has non-bypassable padding, so this padding

            // is bypassable.

        } else {

            // Nobody is blocking right now; it's time to unblock.

        }

}

#[allow(clippy::unnecessary_wraps)]

impl<S: SleepProvider> PaddingController<S> {

    /// Report that we've enqueued a non-padding cell for a given hop.

    ///

    /// Return a QueuedCellPaddingInfo if we need to alert the padding subsystem

    /// when this cell is flushed.

        // Every hop up to and including the target hop will see this as normal data.

    /// Install the given [`CircuitPadder`] to start padding traffic to the listed `hop`.

    ///

    /// Stops padding if the provided padder is `None`.

    ///

    /// Replaces any previous [`CircuitPadder`].

    /// Report that we have enqueued a non-padding cell

    /// in place of a replaceable padding cell

    /// for a given hop.

    ///

    /// Return a QueuedCellPaddingInfo if we need to alert the padding subsystem

    /// when this cell is flushed.

    /// Report that we have enqueued a padding cell to a given hop.

    ///

    /// Return a QueuedCellPaddingInfo if we need to alert the padding subsystem

    /// when this cell is flushed.

    /// Report that we are using an already-queued cell

    /// as a substitute for sending padding to a given hop.

    /// Report that we've flushed a cell from the queue for the given hop.

        // Every hop up to the last

    /// Report that we've flushed a cell from the per-channel queue.

    /// Report that we have decrypted a non-padding cell from our queue

    /// from a given hop.

    ///

    // Note that in theory, it would be better to trigger TunnelRecv as soon as

    // possible after we receive and enqueue the data cell, and NormalRecv only

    // once we've decrypted it and found it to be data.  But we can't do that,

    // since we won't know which hop actually originated the cell until we

    // decrypt it.

    /// Report that we have decrypted a padding cell from our queue.

    ///

    /// Return an error if this padding cell is not acceptable

    /// (because we have received too much padding from this hop,

    /// or because we have not enabled padding with this hop.)

    //

    // See note above.

}

impl<S: SleepProvider> PaddingShared<S> {

    /// Trigger a list of maybenot events on every hop up to and including `hop`.

            };

        }

    /// Trigger `intermediate_hop_events` on every hop up to but _not_ including `hop`.

    ///

    /// Trigger `final_hop_events` on `hop`.

    ///

    /// (Don't trigger anything on any hops _after_ `hop`.)

        use itertools::Itertools as _;

        use itertools::Position as P;

            };

            };

        }

    /// Increment the normal cell count from every hop up to and including `hop`.

    /// Increment the padding count from `hop`, and the normal cell count from all earlier hops.

    ///

    /// Return an error if a padding cell from `hop` would not be acceptable.

        use itertools::Itertools as _;

        use itertools::Position as P;

                }

                (P::Last | P::Only, None) => {

                }

            }

        }

    /// Return the `QueuedCellPaddingInfo` to use when sending messages to `target_hop`

    #[allow(clippy::unnecessary_wraps)]

        // TODO circpad optimization: This is always Some for now, but we

        // could someday avoid creating this object

        // when padding is not enabled on the circuit,

        // or if padding is not enabled on any hop of the circuit <= target_hop.

}

impl<S: SleepProvider> PaddingShared<S> {

    /// Install or remove a [`CircuitPadder`] for a single hop.

        // Make sure there are enough spaces in self.hops.

        // We can't use "resize" or "extend", since Box<dyn<PaddingBackend>>

        // doesn't implement Clone, which SmallVec requires.

        // project through option...

        } else {

        };

        // We need to alert the stream, in case we added an event above, and so that it will poll

        // the new padder at least once.

    /// Transform a [`PerHopPaddingEvent`] for a single hop with index `idx` into a [`PaddingEvent`],

    /// updating our state as appropriate.

        use PaddingEvent as PE;

        use PerHopPaddingEvent as PHPE;

            PHPE::SendPadding {

                // NOTE that we remember is_bypassable for every hop, but the blocking is only

                // bypassable if _every_ hop is unblocked, or has bypassable blocking.

            }

            PHPE::StopBlocking => {

            }

        }

    /// Extract every PaddingEvent that is ready to be reported to the circuit at time `now`.

    ///

    /// May trigger other events, or wake up the stream, in the course of running.

            };

            );

        }

    /// Find the next time at which we should wake up the stream, and register it as our

    /// "next scheduled wakeup".

        // Find the earliest time at which any hop has a scheduled event.

}

/// A stream of [`PaddingEvent`] to tell a circuit when (if at all) it should send

/// padding and block traffic.

//

// TODO circpad: Optimize this even more for the no-padding case?

// We could make it smaller or faster.

pub(crate) struct PaddingEventStream<S = DynTimeProvider>

where

    S: SleepProvider,

{

    /// An underlying list of PaddingBackend.

    shared: Arc<Mutex<PaddingShared<S>>>,

    /// A future defining a time at which we must next call `padder.padding_events_at`.

    ///

    /// (We also arrange for the backend to wake us up if we need to change this time,

    /// or call `padder.padding_events_at`.)

    ///

    /// Note that this timer is allowed to be _earlier_ than our true wakeup time,

    /// but not later.

    sleep_future: S::SleepFuture,

}

impl futures::Stream for PaddingEventStream {

    type Item = PaddingEvent;

        loop {

                // We destructure like this to avoid simultaneous mutable/immutable borrows.

                // Do we have any events that are waiting to be yielded?

                // Does the padder have any events that have become ready to be yielded?

                // If we reach this point, there are no events to trigger right now.

                //

                // We'll ask all the padders for the time at which they next might need to take

                // action, and register our Waker with them, to be alerted if we need to take any action

                // before that.

                // Here we drop the lock on the shared state.

            };

                None => {

                }

                    // TODO circpad: Avoid rebuilding sleep future needlessly.  May require new APIs in

                    // tor-rtcompat.

                        Poll::Ready(()) => {

                            // Okay, The timer expired already. Continue through the loop.

                        }

                    }

                }

            }

        }

}

impl futures::stream::FusedStream for PaddingEventStream {

        // This stream is _never_ terminated: even if it has no padding machines now,

        // we might add some in the future.

}

/// Construct a HopNum from an index into the `hops` field of a [`PaddingShared`].

///

/// # Panics

///

/// Panics if `hop_idx` is greater than u8::MAX, which should be impossible.

    // (Static assertion: makes sure we can represent every index of hops as a HopNum.)

    const _: () = assert!(MAX_HOPS < u8::MAX as usize);

/// Create a new, empty padding instance for a new circuit.

{

    // Start with an arbitrary sleep future.  We won't actually use this until

    // the first time that we have an event to schedule, so the timeout doesn't matter.