Struct tor_hsservice::publish::reactor::Reactor

pub(super) struct Reactor<R: Runtime, M: Mockable> {
    imm: Arc<Immutable<R, M>>,
    dir_provider: Arc<dyn NetDirProvider>,
    inner: Arc<Mutex<Inner>>,
    ipt_watcher: IptsPublisherView,
    config_rx: Receiver<Arc<OnionServiceConfig>>,
    publish_status_rx: Receiver<PublishStatus>,
    publish_status_tx: Sender<PublishStatus>,
    upload_task_complete_rx: Receiver<TimePeriodUploadResult>,
    upload_task_complete_tx: Sender<TimePeriodUploadResult>,
    shutdown_tx: Sender<Void>,
}

A reactor for the HsDir Publisher

The entrypoint is Reactor::run.

Fields

imm: Arc<Immutable<R, M>>

The immutable, shared inner state.

dir_provider: Arc<dyn NetDirProvider>

A source for new network directories that we use to determine our HsDirs.

inner: Arc<Mutex<Inner>>

The mutable inner state,

ipt_watcher: IptsPublisherView

A channel for receiving IPT change notifications.

config_rx: Receiver<Arc<OnionServiceConfig>>

A channel for receiving onion service config change notifications.

publish_status_rx: Receiver<PublishStatus>

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, we can start publishing descriptors.

If the PublishStatus is AwaitingIpts, publishing is paused until we receive a notification on ipt_watcher telling us the IPT manager has established some introduction points.

publish_status_tx: Sender<PublishStatus>

A sender for updating our PublishStatus.

When our PublishStatus changes to UploadScheduled, we can start publishing descriptors.

upload_task_complete_rx: Receiver<TimePeriodUploadResult>

A channel for sending upload completion notifications.

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

upload_task_complete_tx: Sender<TimePeriodUploadResult>

A channel for receiving upload completion notifications.

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

shutdown_tx: Sender<Void>

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

Implementations

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

pub(super) fn new( runtime: R, nickname: HsNickname, dir_provider: Arc<dyn NetDirProvider>, mockable: M, config: Arc<OnionServiceConfig>, ipt_watcher: IptsPublisherView, config_rx: Receiver<Arc<OnionServiceConfig>>, status_tx: PublisherStatusSender, keymgr: Arc<KeyMgr> ) -> Self

Create a new Reactor.

pub(super) async fn run(self) -> Result<(), FatalError>

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.

async fn run_once(&mut self) -> Result<ShutdownStatus, FatalError>

Run one iteration of the reactor loop.

fn status(&self) -> PublishStatus

Returns the current status of the publisher

fn handle_upload_results(&self, results: TimePeriodUploadResult)

Handle a batch of upload outcomes, possibly updating the status of the descriptor for the corresponding HSDirs.

async fn handle_consensus_change( &mut self, netdir: Arc<NetDir> ) -> Result<(), FatalError>

Maybe update our list of HsDirs.

fn recompute_hs_dirs(&self) -> Result<(), FatalError>

Recompute the HsDirs for all relevant time periods.

fn compute_time_periods( &self, netdir: &Arc<NetDir>, time_periods: &[TimePeriodContext] ) -> Result<Vec<TimePeriodContext>, FatalError>

Compute the TimePeriodContexts for the time periods from the specified NetDir.

The specified time_periods are used to preserve the DescriptorStatus of the HsDirs where possible.

fn replace_netdir(&self, new_netdir: Arc<NetDir>) -> Option<Arc<NetDir>>

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

fn replace_config_if_changed(&self, new_config: Arc<OnionServiceConfig>) -> bool

Replace our view of the service config with new_config if new_config contains changes that would cause us to generate a new descriptor.

fn note_ipt_change(&self) -> PublishStatus

Read the intro points from ipt_watcher, and decide whether we’re ready to start uploading.

async fn handle_ipt_change( &mut self, update: Option<Result<(), FatalError>> ) -> Result<ShutdownStatus, FatalError>

Update our list of introduction points.

async fn update_publish_status_unless_waiting( &mut self, new_state: PublishStatus ) -> Result<(), FatalError>

Update the PublishStatus of the reactor with new_state, unless the current state is AwaitingIpts.

async fn update_publish_status_unless_rate_lim( &mut self, new_state: PublishStatus ) -> Result<(), FatalError>

Update the PublishStatus of the reactor with new_state, unless the current state is RateLimited.

async fn update_publish_status( &mut self, new_state: PublishStatus ) -> Result<(), FatalError>

Unconditionally update the PublishStatus of the reactor with new_state.

async fn handle_new_keys(&self) -> Result<(), FatalError>

Use the new keys.

async fn handle_svc_config_change( &mut self, config: Arc<OnionServiceConfig> ) -> Result<(), FatalError>

Update the descriptors based on the config change.

fn mark_all_dirty(&self)

Mark the descriptor dirty for all time periods.

fn mark_dirty(&self, period: &TimePeriod) -> bool

Mark the descriptor dirty for the specified time period.

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

async fn upload_all(&mut self) -> Result<(), FatalError>

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

async fn upload_for_time_period( hs_dirs: Vec<RelayIds>, netdir: &Arc<NetDir>, config: Arc<OnionServiceConfig>, params: HsDirParams, imm: Arc<Immutable<R, M>>, ipt_upload_view: IptsPublisherUploadView, upload_task_complete_tx: Sender<TimePeriodUploadResult>, shutdown_rx: Receiver<Void> ) -> Result<(), FatalError>

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

async fn upload_descriptor( hsdesc: String, netdir: &Arc<NetDir>, hsdir: &Relay<'_>, imm: Arc<Immutable<R, M>> ) -> Result<(), UploadError>

Upload a descriptor to the specified HSDir.

If an upload fails, this returns an Err. This function does not handle retries. It is up to the caller to retry on failure.

This function does not handle timeouts.

async fn upload_descriptor_with_retries( hsdesc: String, netdir: &Arc<NetDir>, hsdir: &Relay<'_>, ed_id: &str, rsa_id: &str, imm: Arc<Immutable<R, M>> ) -> UploadStatus

Upload a descriptor to the specified HSDir, retrying if appropriate.

TODO (#1216): document the retry logic when we implement it.

async fn start_rate_limit(&mut self, delay: Duration) -> Result<(), FatalError>

Stop publishing descriptors until the specified delay elapses.

async fn expire_rate_limit(&mut self) -> Result<(), FatalError>

Handle the upload rate-limit being lifted.