pubhubs/servers/

run.rs

//! Running PubHubs [`Server`]s
use std::num::NonZero;
use std::rc::Rc;
use std::sync::Arc;

use actix_web::web;
use anyhow::{Context as _, Result, bail};
use core::convert::Infallible;
use tokio::sync::mpsc;

use crate::api;
use crate::misc::defer;
use crate::servers::{
    App, AppBase, AppCreator, Command, Constellation, DiscoverVerdict, Name, Server,
    for_all_servers, server::RunningState,
};

/// A set of running PubHubs servers.
pub struct Set {
    /// Handle to the task waiting on [SetInner::wait].
    wait_jh: tokio::task::JoinHandle<usize>,
}

/// Additional options for [`Set::new_opts`].
///
/// Used by the integration tests to run the servers on pre-bound, ephemeral ports, so that
/// multiple [`Set`]s can run simultaneously.  Each listener is `try_clone`d (the file descriptor
/// is dup'd) every time its server (re)starts — e.g. after discovery — so the originals are kept
/// alive (by the `Runner`, for as long as the server runs) to keep the port reserved.
#[derive(Default)]
pub struct SetOpts {
    /// If set, PubHubs Central listens on this pre-bound socket instead of binding the address in
    /// its config.  Its advertised `phc_url` must point at the same port (not checked at runtime).
    pub phc_listener: Option<std::net::TcpListener>,

    /// Like [`SetOpts::phc_listener`], but for the transcryptor.
    pub transcryptor_listener: Option<std::net::TcpListener>,

    /// Like [`SetOpts::phc_listener`], but for the authentication server.
    pub auths_listener: Option<std::net::TcpListener>,
}

impl SetOpts {
    /// Takes the pre-bound listener for the given server, if any.
    fn take_listener(&mut self, name: Name) -> Option<std::net::TcpListener> {
        match name {
            Name::PubhubsCentral => self.phc_listener.take(),
            Name::Transcryptor => self.transcryptor_listener.take(),
            Name::AuthenticationServer => self.auths_listener.take(),
        }
    }
}

impl Set {
    /// Creates a new set of PubHubs servers from the given config.
    /// To signal shutdown of these senders, drop the returned `sender`.
    pub fn new(
        config: &crate::servers::Config,
    ) -> Result<(Self, tokio::sync::oneshot::Sender<Infallible>)> {
        Self::new_opts(config, SetOpts::default())
    }

    /// Like [`Set::new`], but takes additional [`SetOpts`].
    pub fn new_opts(
        config: &crate::servers::Config,
        opts: SetOpts,
    ) -> Result<(Self, tokio::sync::oneshot::Sender<Infallible>)> {
        let (inner, shutdown_sender) = SetInner::new(config, opts)?;

        let wait_jh = tokio::task::spawn(inner.wait());

        Ok((Self { wait_jh }, shutdown_sender))
    }

    /// Waits for one of the servers to return, panic, or be cancelled.
    /// If that happens, the other servers are directed to shutdown as well.
    ///
    /// Returns the number of servers that did *not* shutdown cleanly.
    ///
    /// Panics when the tokio runtime is shut down.
    pub async fn wait(self) -> usize {
        match self.wait_jh.await {
            Ok(nr) => nr,
            Err(join_error) => {
                panic!("task waiting on servers to exit was cancelled or panicked: {join_error}");
            }
        }
    }
}

/// A set of running PubHubs servers.
struct SetInner {
    /// The servers' tasks
    joinset: tokio::task::JoinSet<Result<()>>,

    /// Via `shutdown_sender` [`Set`] broadcasts the instruction to shutdown to all servers
    /// running in the `joinset`.  It does so not by `send`ing a message, but by dropping
    /// the `shutdown_sender`.
    shutdown_sender: Option<tokio::sync::broadcast::Sender<Infallible>>,

    /// Via `shutdown_receiver`, the [Set] received the instruction to close.
    shutdown_receiver: tokio::sync::oneshot::Receiver<Infallible>,
}

impl Drop for SetInner {
    fn drop(&mut self) {
        if self.shutdown_sender.is_some() {
            log::error!(
                "the completion of all pubhubs servers was not awaited - please consume SetInner using wait() or shutdown()"
            )
        }
    }
}

impl SetInner {
    /// Creates a new set of PubHubs servers from the given config.
    ///
    /// Returns not only the [`SetInner`] instance, but also a [`tokio::sync::oneshot::Sender<Infallible>`]
    /// that can be dropped to signal the [`SetInner`] should shutdown.
    pub fn new(
        config: &crate::servers::Config,
        mut opts: SetOpts,
    ) -> Result<(Self, tokio::sync::oneshot::Sender<Infallible>)> {
        let rt_handle: tokio::runtime::Handle = tokio::runtime::Handle::current();
        let mut joinset = tokio::task::JoinSet::<Result<()>>::new();

        let (shutdown_sender, _) = tokio::sync::broadcast::channel(1); // NB capacity of 0 is not allowed

        // count the number of servers
        let server_count: usize = {
            let mut counter: usize = 0;

            macro_rules! count_server {
                ($server:ident) => {
                    if config.$server.is_some() {
                        counter += 1;
                    }
                };
            }

            for_all_servers!(count_server);

            counter
        };

        // don't use one thread per code for each server - this speeds up testing
        let worker_count: Option<NonZero<usize>> =
            std::thread::available_parallelism()
                .ok()
                .map(|parallelism: NonZero<usize>| {
                    if server_count == 0 {
                        return NonZero::<usize>::new(1).unwrap();
                    }

                    NonZero::<usize>::try_from(parallelism.get() / server_count)
                        .unwrap_or(NonZero::<usize>::new(1).unwrap()) // more servers than cores
                });

        macro_rules! run_server {
            ($server:ident) => {
                if config.$server.is_some() {
                    let config = config.clone();
                    let rt_handle = rt_handle.clone();
                    let shutdown_receiver = shutdown_sender.subscribe();
                    let listener =
                        opts.take_listener(<crate::servers::$server::Server as Server>::NAME);

                    // We use spawn_blocking instead of spawn, because we want a separate thread
                    // for each server to run on
                    joinset.spawn_blocking(move || -> Result<()> {
                        Self::run_server::<crate::servers::$server::Server>(
                            config,
                            rt_handle,
                            shutdown_receiver,
                            worker_count,
                            listener,
                        )
                    });
                }
            };
        }

        for_all_servers!(run_server);

        // A pre-bound listener supplied for a server absent from `config` is never taken above and
        // would be dropped silently (releasing its port); catch that misuse in debug builds.
        debug_assert!(
            opts.phc_listener.is_none()
                && opts.transcryptor_listener.is_none()
                && opts.auths_listener.is_none(),
            "SetOpts has a pre-bound listener for a server not present in the config"
        );

        let (external_shutdown_sender, external_shutdown_receiver) =
            tokio::sync::oneshot::channel();

        Ok((
            Self {
                joinset,
                shutdown_sender: Some(shutdown_sender),
                shutdown_receiver: external_shutdown_receiver,
            },
            external_shutdown_sender,
        ))
    }

    // Creates a server from the given `config` and run it ont the given tokio runtime.
    //
    // Abort when the `shutdown_receiver` channel is closed.
    fn run_server<S: Server>(
        config: crate::servers::Config,
        rt_handle: tokio::runtime::Handle,
        shutdown_receiver: tokio::sync::broadcast::Receiver<Infallible>,
        worker_count: Option<NonZero<usize>>,
        listener: Option<std::net::TcpListener>,
    ) -> Result<()> {
        assert!(config.preparation_state == crate::servers::config::PreparationState::Preliminary);

        let localset = tokio::task::LocalSet::new();

        let fut = localset.run_until(async {
            let config = config.prepare_for(S::NAME).await?;

            crate::servers::run::Runner::<S>::new(
                &config,
                shutdown_receiver,
                worker_count,
                listener,
            )?
            .run()
            .await
        });

        let result = rt_handle.block_on(fut);

        rt_handle.block_on(localset);

        log::debug!("{} stopped with {:?}", S::NAME, result);

        result
    }

    /// Waits for one of the servers to return, panic, or be cancelled.
    /// If that happens, the other servers are directed to shutdown as well.
    ///
    /// If this function is not called, servers can fail silently.
    ///
    /// Returns the number of servers that did *not* shutdown cleanly
    pub async fn wait(mut self) -> usize {
        log::trace!("waiting for one of the servers to exit...");
        let err_count: usize = tokio::select! {
            // either one of the servers exits
            result_maybe = self
                .joinset
                .join_next() => {
                    let result = result_maybe.expect("no servers to wait on");
                    let is_err : bool =  matches!(result, Err(_) | Ok(Err(_)));

                    log::log!( if is_err { log::Level::Error } else { log::Level::Debug },
                        "one of the servers exited with {result:?};  stopping all servers.."
                    );

                    if is_err {
                        1
                    } else {
                        0
                    }
                },

            // or we get the command to shut down from higher up
            result = &mut self.shutdown_receiver => {
                result.expect_err("received Infallible");
                log::debug!("shutdown requested"); 0
            }
        };

        self.shutdown().await + err_count
    }

    /// Requests shutdown of all servers, and wait for it to complete.
    /// Returns the number of servers that did *not* shutdown cleanly.
    pub async fn shutdown(mut self) -> usize {
        assert!(
            self.shutdown_sender.is_some(),
            "only signal_shutdown should take shutdown_sender"
        );

        // This causes the shutdown_receivers at the different servers to be closed,
        // which in turn should cause those servers' threads to join.
        drop(self.shutdown_sender.take());

        let mut err_count = 0usize;

        while let Some(result) = self.joinset.join_next().await {
            err_count += if matches!(result, Err(_) | Ok(Err(_))) {
                1
            } else {
                0
            };
        }

        // join_next returned None, which means the JoinSet is empty

        err_count
    }
}

/// Runs a [`Server`]
struct Runner<ServerT: Server> {
    pubhubs_server: Rc<ServerT>,
    shutdown_receiver: tokio::sync::broadcast::Receiver<Infallible>,
    worker_count: Option<NonZero<usize>>,

    /// If set, the server listens on this pre-bound socket — `try_clone`d on each (re)start —
    /// instead of binding the address in its [`ServerConfig`](crate::servers::config::ServerConfig).
    /// Held here so the port stays reserved for the server's whole lifetime.  See [`SetOpts`].
    listener: Option<std::net::TcpListener>,

    /// Number of restarts (i.e. modifications applied)
    generation: usize,
}

/// The handles to control an [actix_web::dev::Server] running a pubhubs [Server].
struct Handles<S: Server> {
    /// Handle to the actual actix TCP server.  The [actix_web::dev::Server] is owned
    /// by the task driving it.
    actix_server_handle: actix_web::dev::ServerHandle,

    /// Handle to the task driving the actix TCP server
    actix_join_handle: tokio::task::JoinHandle<Result<(), std::io::Error>>,

    /// Handle to the task running (discovery for) the PubHubs server.
    ph_join_handle: tokio::task::JoinHandle<Result<Option<crate::servers::server::BoxModifier<S>>>>,

    /// Dropped to order `ph_join_handle` to shutdown.  [None] when used.
    ph_shutdown_sender: Option<tokio::sync::oneshot::Sender<Infallible>>,

    /// Receives commands from the [App]s
    command_receiver: mpsc::Receiver<CommandRequest<S>>,

    /// To check whether [Handles::shutdown] was completed before being dropped
    drop_bomb: crate::misc::drop_ext::Bomb<Box<dyn FnOnce()>>,
}

impl<S: Server> Handles<S> {
    /// Drives the actix server until a [Command] is received - which is returned
    async fn run_until_command(&mut self, runner: &mut Runner<S>) -> anyhow::Result<Command<S>> {
        tokio::select! {

            // received command from running pubhubs/actix server
            command_request_maybe = self.command_receiver.recv() => {
                if let Some(command_request) = command_request_maybe {
                    return Ok(command_request.accept());
                }

                log::error!("{}'s command receiver is unexpectedly closed", S::NAME);
                bail!("{}'s command receiver is unexpectedly closed", S::NAME);
            },

            // pubhubs server exited, returning a modification request
            res = &mut self.ph_join_handle => {
                let modifier : crate::servers::server::BoxModifier<S> =
                res.with_context(|| format!("{}'s pubhubs task joined unexpectedly", S::NAME))?
                    .with_context(|| format!("{}'s pubhubs task crashed", S::NAME))?
                    .with_context(|| format!("{}'s pubhubs task stopped without being asked to", S::NAME))?;


                #[expect(clippy::needless_return)] // "return" makes the code more readable here
                return Ok(Command::Modify(modifier));
            },

            // the thread running this server wants us to quit
            Err(err) = runner.shutdown_receiver.recv() => {
                match err {
                    tokio::sync::broadcast::error::RecvError::Lagged(_) => {
                        panic!("got impossible `Lagged` error from shutdown sender");
                    },
                    tokio::sync::broadcast::error::RecvError::Closed => {
                        #[expect(clippy::needless_return)] // "return" is more readable here
                        return Ok(Command::Exit);
                    },
                }
            },

            // the actix serer exited unexpectedly
            res = &mut self.actix_join_handle => {
                res.inspect_err(|err| log::error!("{}'s actix task joined unexpectedly: {}", S::NAME, err) )
                    .with_context(|| format!("{}'s actix task joined unexpectedly", S::NAME))?
                    .inspect_err(|err| log::error!("{}'s http server crashed: {}", S::NAME, err) )
                    .with_context(|| format!("{}'s http server crashed", S::NAME))?;

                log::error!("{}'s actix server stopped unexpectedly", S::NAME);
                bail!("{}'s actix server stopped unexpectedly", S::NAME);
            },
        };
    }

    /// Consumes this [`Handles`] shutting down the actix server and pubhubs tasks.
    async fn shutdown(mut self) -> anyhow::Result<()> {
        log::debug!("Shut down of {} started", S::NAME);

        anyhow::ensure!(
            self.ph_shutdown_sender.is_some(),
            "shutdown of ph task already ordered"
        );

        drop(self.ph_shutdown_sender.take());

        // This is a noop if the actix server is already stopped
        //
        // We do not use graceful shutdown, becaus in practise the persistent connections
        // delay shutdown for the maximal shutdown timeout, which causes more disruption
        // than the graceful shutdown aims to prevent
        self.actix_server_handle.stop(false).await;

        let maybe_modifier = self
            .ph_join_handle
            .await
            .with_context(|| {
                format!(
                    "{}'s pubhubs task did not join gracefully after being asked to stop",
                    S::NAME
                )
            })?
            .with_context(|| {
                format!(
                    "{}'s pubhubs task crashed after being asked to stop",
                    S::NAME
                )
            })?;

        if let Some(ph_modifier) = maybe_modifier {
            log::error!(
                "Woops! {}'s pubhubs task's modifier {} was ignored, because another modifier was first",
                S::NAME,
                ph_modifier
            );
        }

        self.drop_bomb.defuse();

        log::debug!("Shut down of {} completed", S::NAME);

        Ok(())
    }
}

/// Encapsulates the handling of running just one discovery process per server
struct DiscoveryLimiter {
    /// Lock that makes sure only one discovery task is running at the same time.
    ///
    /// The protected value is true when restart due to a changed constellation is imminent.
    restart_imminent_lock: Arc<tokio::sync::RwLock<bool>>,

    /// Set when contents of `restart_imminent_lock` lock was observed to be true,
    /// reducing the load on this lock.
    restart_imminent_cached: std::cell::OnceCell<()>,
}

impl Clone for DiscoveryLimiter {
    fn clone(&self) -> Self {
        Self {
            restart_imminent_lock: self.restart_imminent_lock.clone(),
            // The DiscoveryLimiter is never cloned as part of an `AppBase`.
            restart_imminent_cached: std::cell::OnceCell::<()>::new(),
        }
    }
}

impl DiscoveryLimiter {
    fn new() -> Self {
        DiscoveryLimiter {
            restart_imminent_lock: Arc::new(tokio::sync::RwLock::new(false)),
            restart_imminent_cached: std::cell::OnceCell::<()>::new(),
        }
    }

    /// This functions contains the discovery logic that's shared between servers.
    ///
    /// Discovery can be invoked for two reasons:  
    ///
    ///  1. This server has just been restarted and is not aware of the current constellation.
    ///     Perhaps part of our configuration has changed that makes the current constellation
    ///     obsolete.
    ///
    ///  2. The `.ph/discovery/run` endpoint was triggered.  This should happen when another
    ///     server detects that our constellation is out-of-date, but since the `.ph/discovery/run`
    ///     endpoint is unprotected, anyone can invoke it at any time.
    ///
    ///     The non-PHC servers check during their discovery whether the constellation PHC advertises
    ///     is up-to-date with respect to their own configuration.  If it isn't, the non-PHC server
    ///     triggers PHC to run discovery.
    ///     
    ///     PHC checks during its discovery (after it obtained recent details from each of the
    ///     other servers) whether the constellations of the other servers are up-to-date,
    ///     and will trigger their discovery routine when these aren't.
    ///
    ///
    /// Thus the procedure for discovery is as follows.
    ///
    ///
    /// Non-PHC:
    ///   
    ///  1. Obtain constellation from PHC.  Return if it coincides with the constellation that we already
    ///     got - if we already got one.
    ///  2. Check the constellation against our own configuration.  If it's up-to-date, restart
    ///     this server but with the new constellation.
    ///  3. Invoke discovery on PHC, and return a retryable error - effectively go back to step 1.
    ///
    ///
    /// PHC:
    ///
    ///  1. Obtain discovery info from ourselves - i.e. check whether `phc_url` is configured
    ///     correctly.
    ///  2. Retrieve discovery info from the other servers and construct a constellation from it.
    ///  3. If the constellation has changed, restart to update it.
    ///  4. Invoke discovery on those servers that have no or outdated constellations,
    ///     and return a retryable error - effectively go back to step 1.
    ///
    ///
    async fn request_discovery<S: Server>(
        &self,
        app: Rc<S::AppT>,
    ) -> api::Result<api::DiscoveryRunResp> {
        log::debug!(
            "{server_name}: discovery is requested",
            server_name = S::NAME
        );

        let mut restart_imminent_guard = match self.obtain_lock().await {
            Some(guard) => guard,
            None => {
                log::debug!(
                    "{server_name}: discovery aborted because the server is already restarting",
                    server_name = S::NAME
                );
                return Ok(api::DiscoveryRunResp::Restarting);
            }
        };

        // Obtain discovery info from PHC (even when we are PHC ourselves, for perhaps
        // the phc_url is misconfigured) and perform some basis checks.
        // Should not return an error when our constellation is out of sync.
        let phc_discovery_info = AppBase::<S>::discover_phc(app.clone()).await?;

        if phc_discovery_info.constellation_or_id.is_none() && S::NAME != Name::PubhubsCentral {
            // PubHubs Central is not yet ready - make the caller retry
            log::info!(
                "Discovery of {} is run but {} has no constellation yet",
                S::NAME,
                Name::PubhubsCentral,
            );
            return Err(api::ErrorCode::PleaseRetry);
        }

        // What discovery determined should happen; threaded into the `modify` closure below.
        enum PostDiscovery<Seed> {
            /// Replace the published constellation and rebuild the running state.
            Republish(Box<Constellation>, Seed),
            /// Keep the published constellation; only rebuild the running state from this seed.
            RebuildRunningState(Seed),
            /// Exit the binary so an updated one is (hopefully) started in its place.
            ExitBinary,
        }

        let post_discovery = match app.discover(phc_discovery_info).await? {
            DiscoverVerdict::ConstellationOutdated {
                new_constellation,
                seed,
            } => PostDiscovery::Republish(new_constellation, seed),
            DiscoverVerdict::RunningStateOutdated { seed } => {
                PostDiscovery::RebuildRunningState(seed)
            }
            DiscoverVerdict::BinaryOutdated => PostDiscovery::ExitBinary,
            DiscoverVerdict::Alright => return Ok(api::DiscoveryRunResp::UpToDate),
        };

        // modify server, and restart (to modify all Apps)

        let display = match &post_discovery {
            PostDiscovery::Republish(..) => "updated constellation after discovery",
            PostDiscovery::RebuildRunningState(_) => "updated running state after discovery",
            PostDiscovery::ExitBinary => "restarting binary hoping to update version",
        };

        let result = app
            .handle
            .modify(display, move |server: &mut S| -> bool {
                let (constellation, seed) = match post_discovery {
                    PostDiscovery::Republish(new_constellation, seed) => (*new_constellation, seed),
                    PostDiscovery::RebuildRunningState(seed) => {
                        let Some(running_state) = server.running_state.as_ref() else {
                            log::error!("{}: running state to rebuild is absent", S::NAME);
                            return false;
                        };
                        (
                            AsRef::<Constellation>::as_ref(&running_state.constellation).clone(),
                            seed,
                        )
                    }
                    PostDiscovery::ExitBinary => {
                        return false; // no, don't restart the server, but exit the binary so that
                        // - hopefully - a new version of the binary will be started by e.g. systemd
                    }
                };

                let extra = match server.create_running_state(&constellation, &seed) {
                    Ok(extra) => extra,
                    Err(err) => {
                        log::error!(
                            "Error while restarting {} after discovery: {}",
                            S::NAME,
                            err
                        );
                        return false; // do not restart
                    }
                };

                let new_url = constellation.url(S::NAME).clone();

                let new_running_state = match RunningState::new(constellation, extra) {
                    Ok(running_state) => running_state,
                    Err(err) => {
                        log::error!(
                            "Error while restarting {} after discovery: {}",
                            S::NAME,
                            err
                        );
                        return false; // do not restart
                    }
                };

                let old_running_state = server.running_state.replace(new_running_state);

                // See if our url has changed
                if old_running_state.is_none_or(|rs| rs.constellation.url(S::NAME) != &new_url) {
                    log::info!("{}: at {}", S::NAME, new_url);
                }

                true // yes, restart this server
            })
            .await;

        if let Err(()) = result {
            log::warn!(
                "failed to initiate restart of {} for discovery, probably because the server is already shutting down",
                S::NAME,
            );
            return Err(api::ErrorCode::PleaseRetry);
        }

        log::trace!(
            "{server_name}: registering imminent restart",
            server_name = S::NAME
        );
        *restart_imminent_guard = true;
        let _ = self.restart_imminent_cached.set(());

        Ok(api::DiscoveryRunResp::Restarting)
    }

    /// Obtains write lock to `self.restart_imminent_lock` when restart is not imminent.
    async fn obtain_lock(&self) -> Option<tokio::sync::RwLockWriteGuard<'_, bool>> {
        if self.restart_imminent_cached.get().is_some() {
            log::trace!("restart imminent: cached");
            return None;
        }

        if *self.restart_imminent_lock.read().await {
            log::trace!("restart imminent: discovered after obtaining read lock");
            let _ = self.restart_imminent_cached.set(());
            return None;
        }

        let restart_imminent_guard = self.restart_imminent_lock.write().await;

        if *restart_imminent_guard {
            // while we re-obtained the lock, discovery has completed
            log::trace!("restart imminent: discovered after obtaining write lock");
            let _ = self.restart_imminent_cached.set(());
            return None;
        }

        Some(restart_imminent_guard)
    }
}

/// Handle to a [`Server`] passed to [`App`]s.
///
/// Used to issue commands to the server.  Since discovery is requested a often a separate struct
/// is used to deal with discovery requests.
pub struct Handle<S: Server> {
    /// To send commands to the server
    sender: mpsc::Sender<CommandRequest<S>>,

    /// To coordinate the handling of discovery requests
    discovery_limiter: DiscoveryLimiter,
}

struct CommandRequest<S: Server> {
    /// The actual command
    command: Command<S>,

    /// A way for the [`Server`] to inform the [`App`] that the command is about to be executed.
    feedback_sender: tokio::sync::oneshot::Sender<()>,
}

impl<S: Server> CommandRequest<S> {
    /// Let's the issuer of the command know that the command is to be fulfilled
    fn accept(self) -> Command<S> {
        if self.feedback_sender.send(()).is_err() {
            log::warn!(
                "The app issuing command '{}' that is about to execute has already dropped.",
                &self.command
            );
        }
        self.command
    }
}

// We cannot use "derive(Clone)", because Server is not Clone.
impl<S: Server> Clone for Handle<S> {
    fn clone(&self) -> Self {
        Handle {
            sender: self.sender.clone(),
            discovery_limiter: self.discovery_limiter.clone(),
        }
    }
}

impl<S: Server> Handle<S> {
    /// Issues command to [Runner].  Waits for the command to be next in line,
    /// but does not wait for the command to be completed.
    ///
    /// May return `Err(())` when another command shutdown the server before this
    /// command could be executed.
    ///
    /// When `Ok(())` is returned, this means the command is guaranteed to be executed momentarily.
    pub(crate) async fn issue_command(&self, command: Command<S>) -> Result<(), ()> {
        let (feedback_sender, feedback_receiver) = tokio::sync::oneshot::channel();

        let result = self
            .sender
            .send(CommandRequest {
                command,
                feedback_sender,
            })
            .await;

        if let Err(send_error) = result {
            log::warn!(
                "{server_name}: since the command receiver is closed (probably because the server is shutting down/restarting) we could not issue the command {cmd:?}",
                server_name = S::NAME,
                cmd = send_error.0.command.to_string(),
            );
            return Err(());
        };

        // Wait for the command to be the next in line.
        //
        // If feedback receiver returns an error, the command might not have executed.
        feedback_receiver.await.map_err(|_| ())
    }

    pub async fn modify(
        &self,
        display: impl std::fmt::Display + Send + 'static,
        modifier: impl FnOnce(&mut S) -> bool + Send + 'static,
    ) -> Result<(), ()> {
        self.issue_command(Command::Modify(Box::new((modifier, display))))
            .await
    }

    /// Executes `inspector` on the server instance, returning its result.
    ///
    /// Returns Err(()) when the command or its result could not be sent, probably because the
    /// server was shutting down.
    pub async fn inspect<T: Send + 'static>(
        &self,
        display: impl std::fmt::Display + Send + 'static,
        inspector: impl FnOnce(&S) -> T + Send + 'static,
    ) -> Result<T, ()> {
        let (sender, receiver) = tokio::sync::oneshot::channel::<T>();

        self.issue_command(Command::Inspect(Box::new((
            |server: &S| {
                if sender.send(inspector(server)).is_err() {
                    log::warn!(
                        "{}: could not return result of inspection because receiver was already closed",
                        S::NAME
                    );
                    // Might happen when the server is restarted ungracefully - so don't panic here.
                }
            },
            display,
        ))))
        .await?;

        receiver.await.map_err(|_| {
            log::warn!(
                "{server_name}: could receive result of inspector",
                server_name = S::NAME,
            );
        })
    }

    pub async fn request_discovery(&self, app: Rc<S::AppT>) -> api::Result<api::DiscoveryRunResp> {
        self.discovery_limiter.request_discovery::<S>(app).await
    }
}

impl<S: Server> Runner<S> {
    pub fn new(
        global_config: &crate::servers::Config,
        shutdown_receiver: tokio::sync::broadcast::Receiver<Infallible>,
        worker_count: Option<NonZero<usize>>,
        listener: Option<std::net::TcpListener>,
    ) -> Result<Self> {
        log::trace!("{}: creating runner...", S::NAME);

        let pubhubs_server = Rc::new(S::new(global_config)?);

        log::trace!("{}: created runner", S::NAME);

        Ok(Runner {
            pubhubs_server,
            shutdown_receiver,
            worker_count,
            listener,
            generation: 0,
        })
    }

    fn create_actix_server(&self) -> Result<Handles<S>> {
        let app_creator: S::AppCreatorT = self.pubhubs_server.deref().clone();

        let (command_sender, command_receiver) = mpsc::channel(1);

        let handle = Handle::<S> {
            sender: command_sender,
            discovery_limiter: DiscoveryLimiter::new(),
        };

        let ac_context: <S::AppCreatorT as AppCreator<S>>::ContextT = Default::default();

        let server_config = self.pubhubs_server.server_config();

        let bind_target: String = match self.listener.as_ref() {
            // Pre-bound socket — the integration tests use this for ephemeral ports.
            Some(listener) => format!("pre-bound socket {:?}", listener.local_addr()),
            None => format!(
                "{}, port {}",
                server_config
                    .ips
                    .iter()
                    .map(|ip| ip.to_string())
                    .collect::<Box<[String]>>()
                    .join(", "),
                server_config.port,
            ),
        };

        log::info!(
            "{}:  binding actix server to {}, running on {:?}",
            S::NAME,
            bind_target,
            std::thread::current().id()
        );

        let app_creator2 = app_creator.clone();
        let handle2 = handle.clone();
        let ac_context2 = ac_context.clone();
        let generation: usize = self.generation;

        let actual_actix_server: actix_web::dev::Server = {
            // Build actix server
            let mut builder: actix_web::HttpServer<_, _, _, _> =
                actix_web::HttpServer::new(move || {
                    let app: Rc<S::AppT> = Rc::new(app_creator2.clone().into_app(
                        &handle2,
                        &ac_context2,
                        generation,
                    ));

                    actix_web::App::new().wrap(S::cors()).configure(
                        |sc: &mut web::ServiceConfig| {
                            // first configure endpoints common to all servers
                            AppBase::<S>::configure_actix_app(&app, sc);

                            // and then server-specific endpoints
                            app.configure_actix_app(sc);

                            let weak = Rc::downgrade(&app);

                            tokio::task::spawn_local(async move {
                                let thread = std::thread::current();

                                log::debug!(
                                    "{}: spawned local task on thread {:?}",
                                    S::NAME,
                                    thread.id()
                                );

                                let _deferred = defer(|| {
                                    log::debug!(
                                        "{}: local task on thread {:?} stopped",
                                        S::NAME,
                                        thread.id()
                                    );
                                });

                                S::AppT::local_task(weak).await;
                            });
                        },
                    )
                })
                .disable_signals(); // we handle signals ourselves

            builder = match self.listener.as_ref() {
                // Reuse the pre-bound listener (the integration tests use this for ephemeral
                // ports), dup'd each restart so the held original keeps the port reserved.
                // See [`SetOpts`].
                Some(listener) => {
                    builder.listen(listener.try_clone().context("cloning pre-bound listener")?)?
                }
                None => builder.bind(server_config)?,
            };

            if let Some(worker_count) = self.worker_count {
                builder = builder.workers(worker_count.get());
            }

            builder.run()
        };

        // start actix server
        let actix_server_handle = actual_actix_server.handle().clone();
        let actix_join_handle = tokio::task::spawn(actual_actix_server);

        // start PH server task (doing discovery)
        let (ph_shutdown_sender, ph_shutdown_receiver) = tokio::sync::oneshot::channel();
        let ph_join_handle =
            tokio::task::spawn_local(self.pubhubs_server.clone().run_until_modifier(
                ph_shutdown_receiver,
                Rc::new(app_creator.into_app(&handle, &ac_context, generation)),
            ));

        Ok(Handles {
            actix_server_handle,
            actix_join_handle,
            command_receiver,
            ph_join_handle,
            ph_shutdown_sender: Some(ph_shutdown_sender),
            drop_bomb: crate::misc::drop_ext::Bomb::panic(|| {
                format!("Part of {} was not shut down properly", S::NAME)
            }),
        })
    }

    pub async fn run(mut self) -> Result<()> {
        loop {
            let modifier = self.run_until_modifier().await?;

            let pubhubs_server_mutref: &mut S =
                Rc::get_mut(&mut self.pubhubs_server).expect("pubhubs_server is still borrowed");

            let modifier_fmt = format!("{modifier}"); // so modifier can be consumed

            log::info!("{}: applying modification {:?}", S::NAME, modifier_fmt);

            if !modifier.modify(pubhubs_server_mutref) {
                log::info!(
                    "{}: not restarting upon request of {:?}",
                    S::NAME,
                    modifier_fmt
                );
                return Ok(());
            }

            self.generation += 1;
            log::info!("{}: restarting...", S::NAME);
        }
    }

    pub async fn run_until_modifier(
        &mut self,
    ) -> Result<crate::servers::server::BoxModifier<S>, anyhow::Error> {
        let mut handles = self.create_actix_server()?;

        let result = Self::run_until_modifier_inner(&mut handles, self).await;

        handles.shutdown().await?;

        result
    }

    async fn run_until_modifier_inner(
        handles: &mut Handles<S>,
        runner: &mut Runner<S>,
    ) -> Result<crate::servers::server::BoxModifier<S>, anyhow::Error> {
        loop {
            match handles.run_until_command(runner).await? {
                Command::Modify(modifier) => {
                    log::debug!(
                        "Stopping {} for modification {:?}...",
                        S::NAME,
                        modifier.to_string()
                    );

                    return Ok::<_, anyhow::Error>(modifier);
                }
                Command::Inspect(inspector) => {
                    log::debug!(
                        "{}: applying inspection {:?}",
                        S::NAME,
                        inspector.to_string()
                    );
                    inspector.inspect(&*runner.pubhubs_server);
                }
                Command::Exit => {
                    log::debug!("Stopping {}, as requested", S::NAME);

                    return Ok::<_, anyhow::Error>(Box::new(crate::servers::server::Exiter));
                }
            }
        }
    }
}