//! The onion service publisher reactor.

//!

//! TODO (#1216): write the docs

//!

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

//! the following state transitions are possible:

//!

//!

//! ```ignore

//!

//!                 update_publish_status(UploadScheduled|AwaitingIpts|RateLimited) +---------------+

//!                +--------------------------------------------------------------->| Bootstrapping |

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

//! +----------+   | update_publish_status(Idle)        +---------+                         |

//! | Shutdown |-- +----------------------------------->| Running |----+                    |

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

//!                |                                                   |                    |

//!                |                                                   |                    |

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

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

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

//! ```

//!

//! Ideally, the publisher should also set the

//! [`OnionServiceStatus`] to `Recovering` whenever a transient

//! upload error occurs, but this is currently not possible:

//!

//!   * making the upload tasks set the status to `Recovering` (on failure) and `Running` (on

//!     success) wouldn't work, because the upload tasks run in parallel (they would race with each

//!     other, and the final status (`Recovering`/`Running`) would be the status of the last upload

//!     task, rather than the real status of the publisher

//!   * making the upload task set the status to `Recovering` on upload failure, and letting

//!     `upload_publish_status` reset it back to `Running also would not work:

//!     `upload_publish_status` sets the status back to `Running` when the publisher enters its

//!     `Idle` state, regardless of the status of its upload tasks

//!

//! TODO: Indeed, setting the status to `Recovering` _anywhere_ would not work, because

//! `upload_publish_status` will just overwrite it. We would need to introduce some new

//! `PublishStatus` variant (currently, the publisher only has 3 states, `Idle`, `UploadScheduled`,

//! `AwaitingIpts`), for the `Recovering` (retrying a failed upload) and `Broken` (the upload

//! failed and we've given up) states. However, adding these 2 new states is non-trivial:

//!

//!   * how do we define "failure"? Is it the failure to upload to a single HsDir, or the failure

//!     to upload to **any** HsDirs?

//!   * what should make the publisher transition out of the `Broken`/`Recovering` states? While

//!     `handle_upload_results` can see the upload results for a batch of HsDirs (corresponding to

//!     a time period), the publisher doesn't do any sort of bookkeeping to know if a previously

//!     failed HsDir upload succeeded in a later upload "batch"

//!

//! For the time being, the publisher never sets the status to `Recovering`, and uses the `Broken`

//! status for reporting fatal errors (crashes).

use tor_config::file_watcher::{

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

};

use tor_netdir::DirEvent;

use crate::config::restricted_discovery::{

    DirectoryKeyProviderList, RestrictedDiscoveryConfig, RestrictedDiscoveryKeys,

};

use crate::config::OnionServiceConfigPublisherView;

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>,

}

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

    /// The restricted discovery authorized clients.

    ///

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

    authorized_clients: Arc<Mutex<Option<RestrictedDiscoveryKeys>>>,

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

    status_tx: PublisherStatusSender,

}

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,

}

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

}

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.

        })

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

}

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

#[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),

    /// A descriptor upload timed out before it could complete.

    #[error("descriptor publication timed out")]

    Timeout,

    /// 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)]

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

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

                );

        }

        select_biased! {

            res = self.upload_task_complete_rx.next().fuse() => {

                let Some(upload_res) = res else {

                    return Ok(ShutdownStatus::Terminate);

                };

                self.handle_upload_results(upload_res);

            },

            () = upload_rate_lim.wait_for_earliest(&self.imm.runtime).fuse() => {

                self.expire_rate_limit().await?;

            },

            () = reupload_tracking.wait_for_earliest(&self.imm.runtime).fuse() => {

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

                return Ok(ShutdownStatus::Continue);

            },

            netdir_event = netdir_events.next().fuse() => {

                let Some(netdir_event) = netdir_event else {

                    debug!("netdir event stream ended");

                    return Ok(ShutdownStatus::Terminate);

                };

                if !matches!(netdir_event, DirEvent::NewConsensus) {

                    return Ok(ShutdownStatus::Continue);

                };

                // The consensus changed. Grab a new NetDir.

                let netdir = match self.dir_provider.netdir(Timeliness::Timely) {

                    Ok(y) => y,

                    Err(e) => {

                        error_report!(e, "HS service {}: netdir unavailable. Retrying...", self.imm.nickname);

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

                        wait_for_netdir(self.dir_provider.as_ref(), Timeliness::Timely)

                            .await?

                    }

                };

                let relevant_periods = netdir.hs_all_time_periods();

                self.handle_consensus_change(netdir).await?;

                expire_publisher_keys(

                    &self.imm.keymgr,

                    &self.imm.nickname,

                    &relevant_periods,

            }

            update = self.ipt_watcher.await_update().fuse() => {

                if self.handle_ipt_change(update).await? == ShutdownStatus::Terminate {

                    return Ok(ShutdownStatus::Terminate);

                }

            },

            config = self.config_rx.next().fuse() => {

                let Some(config) = config else {

                    return Ok(ShutdownStatus::Terminate);

                };

                self.handle_svc_config_change(&config).await?;

            },

            res = self.key_dirs_rx.next().fuse() => {

                let Some(event) = res else {

                    return Ok(ShutdownStatus::Terminate);

                };

                while let Some(_ignore) = self.key_dirs_rx.try_recv() {

                    // Discard other events, so that we only reload once.

                }

                self.handle_key_dirs_change(event).await?;

            }

            should_upload = self.publish_status_rx.next().fuse() => {

                let Some(should_upload) = should_upload else {

                    return Ok(ShutdownStatus::Terminate);

                };

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

                if should_upload == PublishStatus::UploadScheduled {

                    self.update_publish_status_unless_waiting(PublishStatus::Idle).await?;

                    self.upload_all().await?;

                }

            }

        }

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

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

        } else {

        }

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

    ///

    /// Any failed uploads are retried (TODO (#1216, #1098): document the retry logic when we

    /// implement it, as well as in what cases this will return an error).

        // Check if we should rate-limit this upload.

                    } else {

                    }

            // This scope exists because rng is not Send, so it needs to fall out of scope before we

            // await anything.

            );

            );

            // Spawn a task to upload the descriptor to all HsDirs of this time period.

            //

            // This task will shut down when the reactor is dropped (i.e. when shutdown_rx is

            // dropped).

                    {

                            time_period

                        );

        }

    /// Upload the descriptor for the specified time period.

    ///

    /// Any failed uploads are retried (TODO (#1216, #1098): document the retry logic when we

    /// implement it, as well as in what cases this will return an error).

    #[allow(clippy::too_many_arguments)] // TODO: refactor

        /// An error returned from an upload future.

        //

        // Exhaustive, because this is a private type.

        enum PublishError {

            /// The upload was aborted because there are no IPTs.

            ///

            /// This happens because of an inevitable TOCTOU race, where after being notified by

            /// the IPT manager that the IPTs have changed (via `self.ipt_watcher.await_update`),

            /// we find out there actually are no IPTs, so we can't build the descriptor.

            ///

            /// This is a special kind of error that interrupts the current upload task, and is

            /// logged at `debug!` level rather than `warn!` or `error!`.

            ///

            /// Ideally, this shouldn't happen very often (if at all).

            #[error("No IPTs")]

            NoIpts,

            /// The reactor has shut down

            #[error("The reactor has shut down")]