//! The onion service publisher reactor.

//!

//! Generates and publishes hidden service descriptors in response to various events.

//!

//! [`Reactor::run`] is the entry-point of the reactor. It starts the reactor,

//! and runs until [`Reactor::run_once`] returns [`ShutdownStatus::Terminate`]

//! or a fatal error occurs. `ShutdownStatus::Terminate` is returned if

//! any of the channels the reactor is receiving events from is closed

//! (i.e. when the senders are dropped).

//!

//! ## Publisher status

//!

//! The publisher has an internal [`PublishStatus`], distinct from its [`State`],

//! which is used for onion service status reporting.

//!

//! The main loop of the reactor reads the current `PublishStatus` from `publish_status_rx`,

//! and responds by generating and publishing a new descriptor if needed.

//!

//! See [`PublishStatus`] and [`Reactor::publish_status_rx`] for more details.

//!

//! ## When do we publish?

//!

//! We generate and publish a new descriptor if

//!   * the introduction points have changed

//!   * the onion service configuration has changed in a meaningful way (for example,

//!     if the `restricted_discovery` configuration or its [`Anonymity`](crate::Anonymity)

//!     has changed. See [`OnionServiceConfigPublisherView`]).

//!   * there is a new consensus

//!   * it is time to republish the descriptor (after we upload a descriptor,

//!     we schedule it for republishing at a random time between 60 minutes and 120 minutes

//!     in the future)

//!

//! ## Onion service status

//!

//! With respect to [`OnionServiceStatus`] reporting,

//! the following state transitions are possible:

//!

//!

//! ```ignore

//!

//!                 update_publish_status(UploadScheduled|AwaitingIpts|RateLimited)

//!                +---------------------------------------+

//!                |                                       |

//!                |                                       v

//!                |                               +---------------+

//!                |                               | Bootstrapping |

//!                |                               +---------------+

//!                |                                       |

//!                |                                       |           uploaded to at least

//!                |  not enough HsDir uploads succeeded   |        some HsDirs from each ring

//!                |         +-----------------------------+-----------------------+

//!                |         |                             |                       |

//!                |         |              all HsDir uploads succeeded            |

//!                |         |                             |                       |

//!                |         v                             v                       v

//!                |  +---------------------+         +---------+        +---------------------+

//!                |  | DegradedUnreachable |         | Running |        |  DegradedReachable  |

//! +----------+   |  +---------------------+         +---------+        +---------------------+

//! | Shutdown |-- |         |                           |                        |

//! +----------+   |         |                           |                        |

//!                |         |                           |                        |

//!                |         |                           |                        |

//!                |         +---------------------------+------------------------+

//!                |                                     |   invalid authorized_clients

//!                |                                     |      after handling config change

//!                |                                     |

//!                |                                     v

//!                |     run_once() returns an error +--------+

//!                +-------------------------------->| Broken |

//!                                                  +--------+

//! ```

//!

//! We can also transition from `Broken`, `DegradedReachable`, or `DegradedUnreachable`

//! back to `Bootstrapping` (those transitions were omitted for brevity).

use tor_config::file_watcher::{

    self, Event as FileEvent, FileEventReceiver, FileEventSender, FileWatcher, FileWatcherBuilder,

};

use tor_config_path::{CfgPath, CfgPathResolver};

use tor_netdir::{DirEvent, NetDir};

use crate::config::restricted_discovery::{

    DirectoryKeyProviderList, RestrictedDiscoveryConfig, RestrictedDiscoveryKeys,

};

use crate::config::OnionServiceConfigPublisherView;

use crate::status::{DescUploadRetryError, Problem};

use super::*;

// TODO-CLIENT-AUTH: perhaps we should add a separate CONFIG_CHANGE_REPUBLISH_DEBOUNCE_INTERVAL

// for rate-limiting the publish jobs triggered by a change in the config?

//

// Currently the descriptor publish tasks triggered by changes in the config

// are rate-limited via the usual rate limiting mechanism

// (which rate-limits the uploads for 1m).

//

// I think this is OK for now, but we might need to rethink this if it becomes problematic

// (for example, we might want an even longer rate-limit, or to reset any existing rate-limits

// each time the config is modified).

/// The upload rate-limiting threshold.

///

/// Before initiating an upload, the reactor checks if the last upload was at least

/// `UPLOAD_RATE_LIM_THRESHOLD` seconds ago. If so, it uploads the descriptor to all HsDirs that

/// need it. If not, it schedules the upload to happen `UPLOAD_RATE_LIM_THRESHOLD` seconds from the

/// current time.

//

// TODO: We may someday need to tune this value; it was chosen more or less arbitrarily.

const UPLOAD_RATE_LIM_THRESHOLD: Duration = Duration::from_secs(60);

/// The maximum number of concurrent upload tasks per time period.

//

// TODO: this value was arbitrarily chosen and may not be optimal.  For now, it

// will have no effect, since the current number of replicas is far less than

// this value.

//

// The uploads for all TPs happen in parallel.  As a result, the actual limit for the maximum

// number of concurrent upload tasks is multiplied by a number which depends on the TP parameters

// (currently 2, which means the concurrency limit will, in fact, be 32).

//

// We should try to decouple this value from the TP parameters.

const MAX_CONCURRENT_UPLOADS: usize = 16;

/// The maximum time allowed for uploading a descriptor to a single HSDir,

/// across all attempts.

pub(crate) const OVERALL_UPLOAD_TIMEOUT: Duration = Duration::from_secs(5 * 60);

/// A reactor for the HsDir [`Publisher`]

///

/// The entrypoint is [`Reactor::run`].

#[must_use = "If you don't call run() on the reactor, it won't publish any descriptors."]

pub(super) struct Reactor<R: Runtime, M: Mockable> {

    /// The immutable, shared inner state.

    imm: Arc<Immutable<R, M>>,

    /// A source for new network directories that we use to determine

    /// our HsDirs.

    dir_provider: Arc<dyn NetDirProvider>,

    /// The mutable inner state,

    inner: Arc<Mutex<Inner>>,

    /// A channel for receiving IPT change notifications.

    ipt_watcher: IptsPublisherView,

    /// A channel for receiving onion service config change notifications.

    config_rx: watch::Receiver<Arc<OnionServiceConfig>>,

    /// A channel for receiving restricted discovery key_dirs change notifications.

    key_dirs_rx: FileEventReceiver,

    /// A channel for sending restricted discovery key_dirs change notifications.

    ///

    /// A copy of this sender is handed out to every `FileWatcher` created.

    key_dirs_tx: FileEventSender,

    /// A channel for receiving updates regarding our [`PublishStatus`].

    ///

    /// The main loop of the reactor watches for updates on this channel.

    ///

    /// When the [`PublishStatus`] changes to [`UploadScheduled`](PublishStatus::UploadScheduled),

    /// we can start publishing descriptors.

    ///

    /// If the [`PublishStatus`] is [`AwaitingIpts`](PublishStatus::AwaitingIpts), publishing is

    /// paused until we receive a notification on `ipt_watcher` telling us the IPT manager has

    /// established some introduction points.

    publish_status_rx: watch::Receiver<PublishStatus>,

    /// A sender for updating our [`PublishStatus`].

    ///

    /// When our [`PublishStatus`] changes to [`UploadScheduled`](PublishStatus::UploadScheduled),

    /// we can start publishing descriptors.

    publish_status_tx: watch::Sender<PublishStatus>,

    /// A channel for sending upload completion notifications.

    ///

    /// This channel is polled in the main loop of the reactor.

    upload_task_complete_rx: mpsc::Receiver<TimePeriodUploadResult>,

    /// A channel for receiving upload completion notifications.

    ///

    /// A copy of this sender is handed to each upload task.

    upload_task_complete_tx: mpsc::Sender<TimePeriodUploadResult>,

    /// A sender for notifying any pending upload tasks that the reactor is shutting down.

    ///

    /// Receivers can use this channel to find out when reactor is dropped.

    ///

    /// This is currently only used in [`upload_for_time_period`](Reactor::upload_for_time_period).

    /// Any future background tasks can also use this channel to detect if the reactor is dropped.

    ///

    /// Closing this channel will cause any pending upload tasks to be dropped.

    shutdown_tx: broadcast::Sender<Void>,

    /// Path resolver for configuration files.

    path_resolver: Arc<CfgPathResolver>,

    /// Queue on which we receive messages from the [`PowManager`] telling us that a seed has

    /// rotated and thus we need to republish the descriptor for a particular time period.

    update_from_pow_manager_rx: mpsc::Receiver<TimePeriod>,

}

/// The immutable, shared state of the descriptor publisher reactor.

#[derive(Clone)]

struct Immutable<R: Runtime, M: Mockable> {

    /// The runtime.

    runtime: R,

    /// Mockable state.

    ///

    /// This is used for launching circuits and for obtaining random number generators.

    mockable: M,

    /// The service for which we're publishing descriptors.

    nickname: HsNickname,

    /// The key manager,

    keymgr: Arc<KeyMgr>,

    /// A sender for updating the status of the onion service.

    status_tx: PublisherStatusSender,

    /// Proof-of-work state.

    pow_manager: Arc<PowManager<R>>,

}

impl<R: Runtime, M: Mockable> Immutable<R, M> {

    /// Create an [`AesOpeKey`] for generating revision counters for the descriptors associated

    /// with the specified [`TimePeriod`].

    ///

    /// If the onion service is not running in offline mode, the key of the returned `AesOpeKey` is

    /// the private part of the blinded identity key. Otherwise, the key is the private part of the

    /// descriptor signing key.

    ///

    /// Returns an error if the service is running in offline mode and the descriptor signing

    /// keypair of the specified `period` is not available.

    //

    // TODO (#1194): we don't support "offline" mode (yet), so this always returns an AesOpeKey

    // built from the blinded id key

            }

            None => {

                // TODO (#1194): we don't support externally provisioned keys (yet), so this branch

                // is unreachable (for now).

                    // TODO (#1194): internal! is not the right type for this error (we need an

                    // error type for the case where a hidden service running in offline mode has

                    // run out of its pre-previsioned keys).

                    //

                    // This will be addressed when we add support for offline hs_id mode

            }

        };

    /// Generate a revision counter for a descriptor associated with the specified

    /// [`TimePeriod`].

    ///

    /// Returns a revision counter generated according to the [encrypted time in period] scheme.

    ///

    /// [encrypted time in period]: https://spec.torproject.org/rend-spec/revision-counter-mgt.html#encrypted-time

        // TODO: in the future, we might want to compute ope_key once per time period (as oppposed

        // to each time we generate a new descriptor), for performance reasons.

        // TODO: perhaps this should be moved to a new HsDirParams::offset_within_sr() function

}

/// Mockable state for the descriptor publisher reactor.

///

/// This enables us to mock parts of the [`Reactor`] for testing purposes.

#[async_trait]

pub(crate) trait Mockable: Clone + Send + Sync + Sized + 'static {

    /// The type of random number generator.

    type Rng: rand::Rng + rand::CryptoRng;

    /// The type of client circuit.

    type ClientCirc: MockableClientCirc;

    /// Return a random number generator.

    fn thread_rng(&self) -> Self::Rng;

    /// Create a circuit of the specified `kind` to `target`.

    async fn get_or_launch_specific<T>(

        &self,

        netdir: &NetDir,

        kind: HsCircKind,

        target: T,

    ) -> Result<Arc<Self::ClientCirc>, tor_circmgr::Error>

    where

        T: CircTarget + Send + Sync;

    /// Return an estimate-based value for how long we should allow a single

    /// directory upload operation to complete.

    ///

    /// Includes circuit construction, stream opening, upload, and waiting for a

    /// response.

    fn estimate_upload_timeout(&self) -> Duration;

}

/// Mockable client circuit

#[async_trait]

pub(crate) trait MockableClientCirc: Send + Sync {

    /// The data stream type.

    type DataStream: AsyncRead + AsyncWrite + Send + Unpin;

    /// Start a new stream to the last relay in the circuit, using

    /// a BEGIN_DIR cell.

    async fn begin_dir_stream(self: Arc<Self>) -> Result<Self::DataStream, tor_proto::Error>;

}

#[async_trait]

impl MockableClientCirc for ClientCirc {

    type DataStream = tor_proto::stream::DataStream;

}

/// The real version of the mockable state of the reactor.

#[derive(Clone, From, Into)]

pub(crate) struct Real<R: Runtime>(Arc<HsCircPool<R>>);

#[async_trait]

impl<R: Runtime> Mockable for Real<R> {

    type Rng = rand::rngs::ThreadRng;

    type ClientCirc = ClientCirc;

    async fn get_or_launch_specific<T>(

        &self,

        netdir: &NetDir,

        kind: HsCircKind,

        target: T,

    ) -> Result<Arc<ClientCirc>, tor_circmgr::Error>

    where

        T: CircTarget + Send + Sync,

        use tor_circmgr::timeouts::Action;

}

/// The mutable state of a [`Reactor`].

struct Inner {

    /// The onion service config.

    config: Arc<OnionServiceConfigPublisherView>,

    /// Watcher for key_dirs.

    ///

    /// Set to `None` if the reactor is not running, or if `watch_configuration` is false.

    ///

    /// The watcher is recreated whenever the `restricted_discovery.key_dirs` change.

    file_watcher: Option<FileWatcher>,

    /// The relevant time periods.

    ///

    /// This includes the current time period, as well as any other time periods we need to be

    /// publishing descriptors for.

    ///

    /// This is empty until we fetch our first netdir in [`Reactor::run`].

    time_periods: Vec<TimePeriodContext>,

    /// Our most up to date netdir.

    ///

    /// This is initialized in [`Reactor::run`].

    netdir: Option<Arc<NetDir>>,

    /// The timestamp of our last upload.

    ///

    /// This is the time when the last update was _initiated_ (rather than completed), to prevent

    /// the publisher from spawning multiple upload tasks at once in response to multiple external

    /// events happening in quick succession, such as the IPT manager sending multiple IPT change

    /// notifications in a short time frame (#1142), or an IPT change notification that's

    /// immediately followed by a consensus change. Starting two upload tasks at once is not only

    /// inefficient, but it also causes the publisher to generate two different descriptors with

    /// the same revision counter (the revision counter is derived from the current timestamp),

    /// which ultimately causes the slower upload task to fail (see #1142).

    ///

    /// Note: This is only used for deciding when to reschedule a rate-limited upload. It is _not_

    /// used for retrying failed uploads (these are handled internally by

    /// [`Reactor::upload_descriptor_with_retries`]).

    last_uploaded: Option<Instant>,

    /// A max-heap containing the time periods for which we need to reupload the descriptor.

    // TODO: we are currently reuploading more than nececessary.

    // Ideally, this shouldn't contain contain duplicate TimePeriods,

    // because we only need to retain the latest reupload time for each time period.

    //

    // Currently, if, for some reason, we upload the descriptor multiple times for the same TP,

    // we will end up with multiple ReuploadTimer entries for that TP,

    // each of which will (eventually) result in a reupload.

    //

    // TODO: maybe this should just be a HashMap<TimePeriod, Instant>

    //

    // See https://gitlab.torproject.org/tpo/core/arti/-/merge_requests/1971#note_2994950

    reupload_timers: BinaryHeap<ReuploadTimer>,

    /// The restricted discovery authorized clients.

    ///

    /// `None`, unless the service is running in restricted discovery mode.

    authorized_clients: Option<Arc<RestrictedDiscoveryKeys>>,

}

/// The part of the reactor state that changes with every time period.

struct TimePeriodContext {

    /// The HsDir params.

    params: HsDirParams,

    /// The HsDirs to use in this time period.

    ///

    // We keep a list of `RelayIds` because we can't store a `Relay<'_>` inside the reactor

    // (the lifetime of a relay is tied to the lifetime of its corresponding `NetDir`. To

    // store `Relay<'_>`s in the reactor, we'd need a way of atomically swapping out both the

    // `NetDir` and the cached relays, and to convince Rust what we're doing is sound)

    hs_dirs: Vec<(RelayIds, DescriptorStatus)>,

    /// The revision counter of the last successful upload, if any.

    last_successful: Option<RevisionCounter>,

    /// The outcome of the last upload, if any.

    upload_results: Vec<HsDirUploadStatus>,

}

impl TimePeriodContext {

    /// Create a new `TimePeriodContext`.

    ///

    /// Any of the specified `old_hsdirs` also present in the new list of HsDirs

    /// (returned by `NetDir::hs_dirs_upload`) will have their `DescriptorStatus` preserved.

                // Check if the HsDir of this result still exists

    /// Recompute the HsDirs for this time period.

                // Have we uploaded the descriptor to thiw relay before? If so, we don't need to

                // reupload it unless it was already dirty and due for a reupload.

                };

    /// Mark the descriptor dirty for all HSDirs of this time period.

    /// Update the upload result for this time period.

}

/// An error that occurs while trying to upload a descriptor.

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

#[non_exhaustive]

pub enum UploadError {

    /// An error that has occurred after we have contacted a directory cache and made a circuit to it.

    #[error("descriptor upload request failed: {}", _0.error)]

    Request(#[from] RequestFailedError),

    /// Failed to establish circuit to hidden service directory

    #[error("could not build circuit to HsDir")]

    Circuit(#[from] tor_circmgr::Error),

    /// Failed to establish stream to hidden service directory

    #[error("failed to establish directory stream to HsDir")]

    Stream(#[source] tor_proto::Error),

    /// An internal error.

    #[error("Internal error")]

    Bug(#[from] tor_error::Bug),

}

define_asref_dyn_std_error!(UploadError);

impl<R: Runtime, M: Mockable> Reactor<R, M> {

    /// Create a new `Reactor`.

    #[allow(clippy::too_many_arguments)]

        /// The maximum size of the upload completion notifier channel.

        ///

        /// The channel we use this for is a futures::mpsc channel, which has a capacity of

        /// `UPLOAD_CHAN_BUF_SIZE + num-senders`. We don't need the buffer size to be non-zero, as

        /// each sender will send exactly one message.

        const UPLOAD_CHAN_BUF_SIZE: usize = 0;

        // Internally-generated instructions, no need for mq.

    /// Start the reactor.

    ///

    /// Under normal circumstances, this function runs indefinitely.

    ///

    /// Note: this also spawns the "reminder task" that we use to reschedule uploads whenever an

    /// upload fails or is rate-limited.

        {

        loop {

                Ok(ShutdownStatus::Terminate) => {

                }

                    );

                }

            }

        }

    /// Run one iteration of the reactor loop.

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

                // We are no longer rate-limited

                // First, extract all the timeouts that already elapsed.

                    // We are not ready to schedule any more reuploads.

                    //

                    // How much we need to sleep is implicitly

                    // tracked in reupload_tracking (through

                    // the TrackingNow implementation)

                }

            }

        }

        // Check if it's time to schedule any reuploads.

                    time_period=?period,

                );

        }

                };

            },

            },

                // Run another iteration, executing run_once again. This time, we will remove the

                // expired reupload from self.reupload_timers, mark the descriptor dirty for all

                // relevant HsDirs, and schedule the upload by setting our status to

                // UploadScheduled.

            },

                };

                // The consensus changed. Grab a new NetDir.

                        // Hopefully a netdir will appear in the future.

                        // in the meantime, suspend operations.

                        //

                        // TODO (#1218): there is a bug here: we stop reading on our inputs

                        // including eg publish_status_rx, but it is our job to log some of

                        // these things.  While we are waiting for a netdir, all those messages

                        // are "stuck"; they'll appear later, with misleading timestamps.

                        //

                        // Probably this should be fixed by moving the logging

                        // out of the reactor, where it won't be blocked.

                    }

                };

            },

                };

            },

                };

            }

                };

                // Our PublishStatus changed -- are we ready to publish?

            }

                };

            }

        }

    /// Returns the current status of the publisher

    /// Handle a batch of upload outcomes,

    /// possibly updating the status of the descriptor for the corresponding HSDirs.

            // The uploads were for a time period that is no longer relevant, so we

            // can ignore the result.

        };

        // We will need to reupload this descriptor at at some point, so we pick

        // a random time between 60 minutes and 120 minutes in the future.

        //

        // See https://spec.torproject.org/rend-spec/deriving-keys.html#WHEN-HSDESC

            time_period=?time_period,

        );

                // This HSDir went away, so the result doesn't matter.

                // Continue processing the rest of the results

            };

                };

        }

    /// Maybe update our list of HsDirs.

        // If the time period has changed, some of our upload results may now be irrelevant,

        // so we might need to update our status (for example, if our uploads are

        // for a no-longer-relevant time period, it means we might be able to update

        // out status from "degraded" to "running")

    /// Recompute the HsDirs for all relevant time periods.

        );

        // Update our list of relevant time periods.

    /// Compute the [`TimePeriodContext`]s for the time periods from the specified [`NetDir`].

    ///

    /// The specified `time_periods` are used to preserve the `DescriptorStatus` of the

    /// HsDirs where possible.

                        // Note: for now, read_blind_id_keypair cannot return Ok(None).

                        // It's supposed to return Ok(None) if we're in offline hsid mode,

                        // but that might change when we do #1194

                // If our previous `TimePeriodContext`s also had an entry for `period`, we need to

                // preserve the `DescriptorStatus` of its HsDirs. This helps prevent unnecessarily

                // publishing the descriptor to the HsDirs that already have it (the ones that are

                // marked with DescriptorStatus::Clean).

                //

                // In other words, we only want to publish to those HsDirs that

                //   * are part of a new time period (which we have never published the descriptor

                //   for), or

                //   * have just been added to the ring of a time period we already knew about

                {

                } else {

                    // Passing an empty iterator here means all HsDirs in this TimePeriodContext

                    // will be marked as dirty, meaning we will need to upload our descriptor to them.

                }

    /// Replace the old netdir with the new, returning the old.

    /// Replace our view of the service config with `new_config` if `new_config` contains changes

    /// that would cause us to generate a new descriptor.

        };

    /// Recreate the FileWatcher for watching the restricted discovery key_dirs.

        }

    /// Read the intro points from `ipt_watcher`, and decide whether we're ready to start

    /// uploading.

        }

    /// Update our list of introduction points.

            Some(Ok(())) => {

            }

            None => {

            }

        }

    /// Update the `PublishStatus` of the reactor with `new_state`,

    /// unless the current state is `AwaitingIpts`.

    /// Update the `PublishStatus` of the reactor with `new_state`,

    /// unless the current state is `RateLimited`.

        // We can't exit this state until the rate-limit expires.

    /// Unconditionally update the `PublishStatus` of the reactor with `new_state`.

            PublishStatus::UploadScheduled

            | PublishStatus::AwaitingIpts

        };

            new_state

        );

    /// Update the onion svc status based on the results of the last descriptor uploads.

    /// Update the descriptors based on the config change.

    /// Update the descriptors based on a restricted discovery key_dirs change.

    ///

    /// If the authorized clients from the [`RestrictedDiscoveryConfig`] have changed,

    /// this marks the descriptor as dirty for all time periods,

    /// and schedules a reupload.

        };

        // Update the file watcher, in case the change was triggered by a key_dir move.

    /// Recreate the authorized_clients based on the current config.

    ///

    /// Returns `true` if the authorized clients have changed.

    /// Read the authorized `RestrictedDiscoveryKeys` from `config`.

            );

    /// Mark the descriptor dirty for all time periods.

    /// Mark the descriptor dirty for the specified time period.

    ///

    /// Returns `true` if the specified period is still relevant, and `false` otherwise.

            }

        }

    /// Try to upload our descriptor to the HsDirs that need it.

    ///

    /// If we've recently uploaded some descriptors, we return immediately and schedule the upload

    /// to happen after [`UPLOAD_RATE_LIM_THRESHOLD`].

    ///

    /// Failed uploads are retried

    /// (see [`upload_descriptor_with_retries`](Reactor::upload_descriptor_with_retries)).

    ///

    /// If restricted discovery mode is enabled and there are no authorized clients,

    /// we abort the upload and set our status to [`State::Broken`].

    //

    // Note: a broken restricted discovery config won't prevent future uploads from being scheduled

    // (for example if the IPTs change),

    // which can can cause the publisher's status to oscillate between `Bootstrapping` and `Broken`.

    // TODO: we might wish to refactor the publisher to be more sophisticated about this.

    //

    /// For each current time period, we spawn a task that uploads the descriptor to

    /// all the HsDirs on the HsDir ring of that time period.

    /// Each task shuts down on completion, or when the reactor is dropped.

    ///

    /// Each task reports its upload results (`TimePeriodUploadResult`)

    /// via the `upload_task_complete_tx` channel.

    /// The results are received and processed in the main loop of the reactor.

    ///

    /// Returns an error if it fails to spawn a task, or if an internal error occurs.

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

        // Abort the upload entirely if we have an empty list of authorized clients

            }

        };