// Copyright 2026 Cloudflare, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! Server process and configuration management

mod bootstrap_services;
pub mod configuration;
#[cfg(unix)]
mod daemon;
#[cfg(unix)]
pub(crate) mod transfer_fd;

use async_trait::async_trait;
#[cfg(unix)]
use daemon::daemonize;
use daggy::NodeIndex;
use log::{error, info, warn};
use parking_lot::Mutex;
#[cfg(all(feature = "dial9", feature = "dial9-worker-s3"))]
pub use pingora_runtime::Dial9S3UploadOpts;
use pingora_runtime::{BlockingPoolOpts, Runtime, RuntimeBuilder};
#[cfg(feature = "dial9")]
pub use pingora_runtime::{
    Dial9RuntimeOpts, DEFAULT_DIAL9_MAX_FILE_SIZE, DEFAULT_DIAL9_MAX_TOTAL_SIZE,
};
pub use pingora_runtime::{RuntimeMetricsOpts, RuntimeOpts};
use pingora_timeout::fast_timeout;
#[cfg(feature = "sentry")]
use sentry::ClientOptions;
use std::sync::Arc;
use std::thread;
use std::time::{Instant, SystemTime};
#[cfg(unix)]
use tokio::signal::unix;
use tokio::sync::{broadcast, watch};
use tokio::time::{sleep, Duration};

use crate::prelude::background_service;
use crate::server::bootstrap_services::{Bootstrap, BootstrapService};
use crate::services::{
    DependencyGraph, ServiceHandle, ServiceReadyNotifier, ServiceReadyWatch, ServiceWithDependents,
};
use configuration::{Opt, ServerConf};
use std::collections::HashMap;
#[cfg(unix)]
pub use transfer_fd::Fds;

use pingora_error::{Error, ErrorType, Result};

/* Time to wait before exiting the program.
This is the graceful period for all existing sessions to finish */
const EXIT_TIMEOUT: u64 = 60 * 5;
/* Time to wait before shutting down listening sockets.
This is the graceful period for the new service to get ready */
const CLOSE_TIMEOUT: u64 = 5;

enum ShutdownType {
    Graceful,
    Quick,
}

/// Internal wrapper for services with dependency metadata.
pub(crate) struct ServiceWrapper {
    ready_notifier: Option<ServiceReadyNotifier>,
    service: Box<dyn ServiceWithDependents>,
    service_handle: ServiceHandle,
}

/// The execution phase the server is currently in.
#[derive(Clone, Debug)]
#[non_exhaustive]
pub enum ExecutionPhase {
    /// The server was created, but has not started yet.
    Setup,

    /// Services are being prepared.
    ///
    /// During graceful upgrades this phase acquires the listening FDs from the old process.
    Bootstrap,

    /// Bootstrap has finished, listening FDs have been transferred.
    BootstrapComplete,

    /// The server is running and is listening for shutdown signals.
    Running,

    /// A QUIT signal was received, indicating that a new process wants to take over.
    ///
    /// The server is trying to send the fds to the new process over a Unix socket.
    GracefulUpgradeTransferringFds,

    /// FDs have been sent to the new process.
    /// Waiting a fixed amount of time to allow the new process to take the sockets.
    GracefulUpgradeCloseTimeout,

    /// A TERM signal was received, indicating that the server should shut down gracefully.
    GracefulTerminate,

    /// The server is shutting down.
    ShutdownStarted,

    /// Waiting for the configured grace period to end before shutting down.
    ShutdownGracePeriod,

    /// Wait for runtimes to finish.
    ShutdownRuntimes,

    /// The server has stopped.
    Terminated,
}

/// The receiver for server's shutdown event. The value will turn to true once the server starts
/// to shutdown
pub type ShutdownWatch = watch::Receiver<bool>;
#[cfg(unix)]
pub type ListenFds = Arc<Mutex<Fds>>;

/// Optional process-owned idle wait used to end a graceful shutdown early.
///
/// The callback must return `true` when all accepted work has become idle and
/// `false` when the supplied deadline expires. The owner is responsible for
/// synchronizing accept quiescence with its activity count before returning
/// `true`; a zero activity count alone is not sufficient.
pub type GracefulShutdownIdle = Arc<dyn Fn(Duration) -> bool + Send + Sync>;

/// Wait for graceful shutdown to become idle, retaining the fixed wait when
/// no idle callback is installed.
fn wait_for_graceful_shutdown_idle(
    idle_wait: Option<&GracefulShutdownIdle>,
    grace_period: Duration,
) -> bool {
    idle_wait.map_or_else(
        || {
            thread::sleep(grace_period);
            false
        },
        |wait_for_idle| wait_for_idle(grace_period),
    )
}

/// The type of shutdown process that has been requested.
#[derive(Debug)]
pub enum ShutdownSignal {
    /// Send file descriptors to the new process before starting runtime shutdown with
    /// [ServerConf::graceful_shutdown_timeout_seconds] timeout.
    GracefulUpgrade,
    /// Wait for [ServerConf::grace_period_seconds] before starting runtime shutdown with
    /// [ServerConf::graceful_shutdown_timeout_seconds] timeout.
    GracefulTerminate,
    /// Shutdown with no timeout for runtime shutdown.
    FastShutdown,
}

/// Watcher of a shutdown signal, e.g., [UnixShutdownSignalWatch] for Unix-like
/// platforms.
#[async_trait]
pub trait ShutdownSignalWatch {
    /// Returns the desired shutdown type once one has been requested.
    async fn recv(&self) -> ShutdownSignal;
}

/// A Unix shutdown watcher that awaits for Unix signals.
///
/// - `SIGQUIT`: graceful upgrade
/// - `SIGTERM`: graceful terminate
/// - `SIGINT`: fast shutdown
#[cfg(unix)]
pub struct UnixShutdownSignalWatch;

#[cfg(unix)]
#[async_trait]
impl ShutdownSignalWatch for UnixShutdownSignalWatch {
    async fn recv(&self) -> ShutdownSignal {
        let mut graceful_upgrade_signal = unix::signal(unix::SignalKind::quit()).unwrap();
        let mut graceful_terminate_signal = unix::signal(unix::SignalKind::terminate()).unwrap();
        let mut fast_shutdown_signal = unix::signal(unix::SignalKind::interrupt()).unwrap();

        tokio::select! {
            _ = graceful_upgrade_signal.recv() => {
                ShutdownSignal::GracefulUpgrade
            },
            _ = graceful_terminate_signal.recv() => {
                ShutdownSignal::GracefulTerminate
            },
            _ = fast_shutdown_signal.recv() => {
                ShutdownSignal::FastShutdown
            },
        }
    }
}

/// Arguments to configure running of the pingora server.
pub struct RunArgs {
    /// Signal for initating shutdown
    #[cfg(unix)]
    pub shutdown_signal: Box<dyn ShutdownSignalWatch>,
}

impl Default for RunArgs {
    #[cfg(unix)]
    fn default() -> Self {
        Self {
            shutdown_signal: Box::new(UnixShutdownSignalWatch),
        }
    }

    #[cfg(windows)]
    fn default() -> Self {
        Self {}
    }
}

/// The server object
///
/// This object represents an entire pingora server process which may have multiple independent
/// services (see [crate::services]). The server object handles signals, reading configuration,
/// zero downtime upgrade and error reporting.
pub struct Server {
    services: HashMap<NodeIndex, ServiceWrapper>,
    shutdown_watch: watch::Sender<bool>,
    // TODO: we many want to drop this copy to let sender call closed()
    shutdown_recv: ShutdownWatch,

    /// Tracks the execution phase of the server during upgrades and graceful shutdowns.
    ///
    /// Users can subscribe to the phase with [`Self::watch_execution_phase()`].
    execution_phase_watch: broadcast::Sender<ExecutionPhase>,

    /// Specification of service level dependencies
    dependencies: Arc<Mutex<DependencyGraph>>,

    /// Service initialization
    bootstrap: Arc<Mutex<Bootstrap>>,

    /// The parsed server configuration
    pub configuration: Arc<ServerConf>,
    /// The parser command line options
    pub options: Option<Opt>,

    /// Optional process-owned idle wait for graceful shutdown.
    graceful_shutdown_idle: Option<GracefulShutdownIdle>,
}

// TODO: delete the pid when exit

impl Server {
    /// Install an optional idle wait for graceful termination.
    ///
    /// When no wait is installed, graceful shutdown retains the historical
    /// fixed grace-period behavior. The callback is invoked immediately after
    /// the shutdown broadcast, without waiting for listeners to stop accepting
    /// new work, and receives the configured grace period as its maximum wait.
    /// The callback owner must synchronize accept quiescence with its activity
    /// count before returning `true`.
    pub fn set_graceful_shutdown_idle(&mut self, idle_wait: GracefulShutdownIdle) {
        self.graceful_shutdown_idle = Some(idle_wait);
    }

    /// Remove a previously installed idle wait and restore the fixed grace period.
    pub fn clear_graceful_shutdown_idle(&mut self) {
        self.graceful_shutdown_idle = None;
    }

    /// Acquire a receiver for the server's execution phase.
    ///
    /// The receiver will produce values for each transition.
    pub fn watch_execution_phase(&self) -> broadcast::Receiver<ExecutionPhase> {
        self.execution_phase_watch.subscribe()
    }

    #[cfg(unix)]
    async fn main_loop(&self, run_args: RunArgs) -> ShutdownType {
        // waiting for exit signal

        self.execution_phase_watch
            .send(ExecutionPhase::Running)
            .ok();

        match run_args.shutdown_signal.recv().await {
            ShutdownSignal::FastShutdown => {
                info!("SIGINT received, exiting");
                ShutdownType::Quick
            }
            ShutdownSignal::GracefulTerminate => {
                // we receive a graceful terminate, all instances are instructed to stop
                info!("SIGTERM received, gracefully exiting");
                // graceful shutdown if there are listening sockets
                info!("Broadcasting graceful shutdown");
                match self.shutdown_watch.send(true) {
                    Ok(_) => {
                        info!("Graceful shutdown started!");
                    }
                    Err(e) => {
                        error!("Graceful shutdown broadcast failed: {e}");
                    }
                }
                info!("Broadcast graceful shutdown complete");

                self.execution_phase_watch
                    .send(ExecutionPhase::GracefulTerminate)
                    .ok();

                ShutdownType::Graceful
            }
            ShutdownSignal::GracefulUpgrade => {
                // TODO: still need to select! on signals in case a fast shutdown is needed
                // aka: move below to another task and only kick it off here
                info!("SIGQUIT received, sending socks and gracefully exiting");

                self.execution_phase_watch
                    .send(ExecutionPhase::GracefulUpgradeTransferringFds)
                    .ok();

                let sent_fds = {
                    let fds = self.listen_fds();
                    let fds = fds.lock();
                    if fds.is_empty() {
                        info!("No socks to send, shutting down.");
                        false
                    } else {
                        info!("Trying to send socks");
                        match fds.send_to_sock(self.configuration.as_ref().upgrade_sock.as_str()) {
                            Ok(_) => {
                                info!("listener sockets sent");
                            }
                            Err(e) => {
                                error!("Unable to send listener sockets to new process: {e}");
                                #[cfg(all(not(debug_assertions), feature = "sentry"))]
                                sentry::capture_error(&e);
                            }
                        }
                        true
                    }
                };
                if sent_fds {
                    self.execution_phase_watch
                        .send(ExecutionPhase::GracefulUpgradeCloseTimeout)
                        .ok();
                    sleep(Duration::from_secs(CLOSE_TIMEOUT)).await;
                }
                info!("Broadcasting graceful shutdown");
                // gracefully exiting
                match self.shutdown_watch.send(true) {
                    Ok(_) => {
                        info!("Graceful shutdown started!");
                    }
                    Err(e) => {
                        error!("Graceful shutdown broadcast failed: {e}");
                        // switch to fast shutdown
                        return ShutdownType::Graceful;
                    }
                }
                info!("Broadcast graceful shutdown complete");
                ShutdownType::Graceful
            }
        }
    }

    #[cfg(windows)]
    async fn main_loop(&self, _run_args: RunArgs) -> ShutdownType {
        // waiting for exit signal

        self.execution_phase_watch
            .send(ExecutionPhase::Running)
            .ok();

        match tokio::signal::ctrl_c().await {
            Ok(()) => {
                info!("Ctrl+C received, gracefully exiting");
                // graceful shutdown if there are listening sockets
                info!("Broadcasting graceful shutdown");
                match self.shutdown_watch.send(true) {
                    Ok(_) => {
                        info!("Graceful shutdown started!");
                    }
                    Err(e) => {
                        error!("Graceful shutdown broadcast failed: {e}");
                    }
                }
                info!("Broadcast graceful shutdown complete");

                self.execution_phase_watch
                    .send(ExecutionPhase::GracefulTerminate)
                    .ok();

                ShutdownType::Graceful
            }
            Err(e) => {
                error!("Unable to listen for shutdown signal: {}", e);
                ShutdownType::Quick
            }
        }
    }

    #[cfg(feature = "sentry")]
    #[cfg_attr(docsrs, doc(cfg(feature = "sentry")))]
    /// The Sentry ClientOptions.
    ///
    /// Panics and other events sentry captures will be sent to this DSN **only in release mode**
    pub fn set_sentry_config(&mut self, sentry_config: ClientOptions) {
        self.bootstrap.lock().set_sentry_config(Some(sentry_config));
    }

    /// Get the configured file descriptors for listening
    #[cfg(unix)]
    fn listen_fds(&self) -> ListenFds {
        self.bootstrap.lock().get_fds()
    }

    /// Collect the listening bind addresses across all registered services.
    #[cfg(unix)]
    fn collect_listen_addresses(&self) -> Option<std::collections::HashSet<String>> {
        self.services
            .values()
            .try_fold(std::collections::HashSet::new(), |mut addrs, wrapper| {
                addrs.extend(wrapper.service.listen_addresses()?);
                Some(addrs)
            })
    }

    #[allow(clippy::too_many_arguments)]
    fn run_service(
        mut service: Box<dyn ServiceWithDependents>,
        #[cfg(unix)] fds: ListenFds,
        shutdown: ShutdownWatch,
        threads: usize,
        work_stealing: bool,
        listeners_per_fd: usize,
        ready_notifier: ServiceReadyNotifier,
        dependency_watches: Vec<ServiceReadyWatch>,
        blocking_opts: BlockingPoolOpts,
        runtime_opts: RuntimeOpts,
    ) -> Runtime
// NOTE: we need to keep the runtime outside async since
        // otherwise the runtime will be dropped.
    {
        let service_runtime = Server::create_runtime(
            service.name(),
            threads,
            work_stealing,
            blocking_opts,
            runtime_opts,
        );
        let service_name = service.name().to_string();
        service_runtime.get_handle().spawn(async move {
            // Wait for all dependencies to be ready
            let mut time_waited_opt: Option<Duration> = None;
            for mut watch in dependency_watches {
                let start = SystemTime::now();

                if watch.wait_for(|&ready| ready).await.is_err() {
                    error!(
                        "Service '{}' dependency channel closed before ready",
                        service_name
                    );
                }

                *time_waited_opt.get_or_insert_default() += start.elapsed().unwrap_or_default()
            }

            if let Some(time_waited) = time_waited_opt {
                service.on_startup_delay(time_waited);
            }

            // Start the actual service, passing the ready notifier
            service
                .start_service(
                    #[cfg(unix)]
                    Some(fds),
                    shutdown,
                    listeners_per_fd,
                    ready_notifier,
                )
                .await;
            info!("service '{}' exited.", service_name);
        });
        service_runtime
    }

    /// Create a new [`Server`], using the [`Opt`] and [`ServerConf`] values provided
    ///
    /// This method is intended for pingora frontends that are NOT using the built-in
    /// command line and configuration file parsing, and are instead using their own.
    ///
    /// If a configuration file path is provided as part of `opt`, it will be ignored
    /// and a warning will be logged.
    pub fn new_with_opt_and_conf(raw_opt: impl Into<Option<Opt>>, mut conf: ServerConf) -> Server {
        let opt = raw_opt.into();
        if let Some(opts) = &opt {
            if let Some(c) = opts.conf.as_ref() {
                warn!("Ignoring command line argument using '{c}' as configuration, and using provided configuration instead.");
            }
            conf.merge_with_opt(opts);
        }

        let (tx, rx) = watch::channel(false);

        let execution_phase_watch = broadcast::channel(100).0;
        let bootstrap = Arc::new(Mutex::new(Bootstrap::new(
            &opt,
            &conf,
            &execution_phase_watch,
        )));

        Server {
            services: Default::default(),
            shutdown_watch: tx,
            shutdown_recv: rx,
            execution_phase_watch,
            configuration: Arc::new(conf),
            options: opt,
            dependencies: Arc::new(Mutex::new(DependencyGraph::new())),
            bootstrap,
            graceful_shutdown_idle: None,
        }
    }

    /// Create a new [`Server`].
    ///
    /// Only one [`Server`] needs to be created for a process. A [`Server`] can hold multiple
    /// independent services.
    ///
    /// Command line options can either be passed by parsing the command line arguments via
    /// `Opt::parse_args()`, or be generated by other means.
    pub fn new(opt: impl Into<Option<Opt>>) -> Result<Server> {
        let opt = opt.into();
        let (tx, rx) = watch::channel(false);

        let execution_phase_watch = broadcast::channel(100).0;
        let conf = if let Some(opt) = opt.as_ref() {
            opt.conf.as_ref().map_or_else(
                || {
                    // options, no conf, generated
                    ServerConf::new_with_opt_override(opt).ok_or_else(|| {
                        Error::explain(ErrorType::ReadError, "Conf generation failed")
                    })
                },
                |_| {
                    // options and conf loaded
                    ServerConf::load_yaml_with_opt_override(opt)
                },
            )
        } else {
            ServerConf::new()
                .ok_or_else(|| Error::explain(ErrorType::ReadError, "Conf generation failed"))
        }?;

        let bootstrap = Arc::new(Mutex::new(Bootstrap::new(
            &opt,
            &conf,
            &execution_phase_watch,
        )));

        Ok(Server {
            services: Default::default(),
            shutdown_watch: tx,
            shutdown_recv: rx,
            execution_phase_watch,
            configuration: Arc::new(conf),
            options: opt,
            dependencies: Arc::new(Mutex::new(DependencyGraph::new())),
            bootstrap,
            graceful_shutdown_idle: None,
        })
    }

    /// Add a service to this server.
    ///
    /// Returns a [`ServiceHandle`] that can be used to declare dependencies.
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// let db_id = server.add_service(database_service);
    /// let api_id = server.add_service(api_service);
    ///
    /// // Declare that API depends on database
    /// api_id.add_dependency(&db_id);
    /// ```
    pub fn add_service(&mut self, service: impl ServiceWithDependents + 'static) -> ServiceHandle {
        self.add_boxed_service(Box::new(service))
    }

    /// Add a pre-boxed service to this server.
    ///
    /// Returns a [`ServiceHandle`] that can be used to declare dependencies.
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// let db_id = server.add_service(database_service);
    /// let api_id = server.add_service(api_service);
    ///
    /// // Declare that API depends on database
    /// api_id.add_dependency(&db_id);
    /// ```
    pub fn add_boxed_service(
        &mut self,
        service_box: Box<dyn ServiceWithDependents>,
    ) -> ServiceHandle {
        let name = service_box.name().to_string();

        // Create a readiness notifier for this service
        let (tx, rx) = watch::channel(false);

        let id = self.dependencies.lock().add_node(name.clone(), rx.clone());

        let service_handle = ServiceHandle::new(id, name, rx, &self.dependencies);

        let wrapper = ServiceWrapper {
            ready_notifier: Some(ServiceReadyNotifier::new(tx)),
            service: service_box,
            service_handle: service_handle.clone(),
        };

        self.services.insert(id, wrapper);

        service_handle
    }

    /// Similar to [`Self::add_service()`], but take a list of services.
    ///
    /// Returns a `Vec<ServiceHandle>` for all added services.
    pub fn add_services(
        &mut self,
        services: Vec<Box<dyn ServiceWithDependents>>,
    ) -> Vec<ServiceHandle> {
        services
            .into_iter()
            .map(|service| self.add_boxed_service(service))
            .collect()
    }

    /// Prepare the server to start
    ///
    /// When trying to zero downtime upgrade from an older version of the server which is already
    /// running, this function will try to get all its listening sockets in order to take them over.
    pub fn bootstrap(&mut self) {
        self.bootstrap.lock().bootstrap();
    }

    /// Create a service that will run to prepare the service to start
    ///
    /// The created service will handle the zero-downtime upgrade from an older version of the server
    /// to this one. It will try to get all its listening sockets in order to take them over.
    pub fn bootstrap_as_a_service(&mut self) -> ServiceHandle {
        let bootstrap_service =
            background_service("Bootstrap Service", BootstrapService::new(&self.bootstrap));

        self.add_service(bootstrap_service)
    }

    /// Start the server using [Self::run] and default [RunArgs].
    ///
    /// This function will block forever until the server needs to quit. So this would be the last
    /// function to call for this object.
    ///
    /// Note: this function may fork the process for daemonization, so any additional threads created
    /// before this function will be lost to any service logic once this function is called.
    pub fn run_forever(self) -> ! {
        self.run(RunArgs::default());

        std::process::exit(0)
    }

    /// Run the server until execution finished.
    ///
    /// This function will run until the server has been instructed to shut down
    /// through a signal, and will then wait for all services to finish and
    /// runtimes to exit.
    ///
    /// Note: if daemonization is enabled in the config, this function will
    /// never return.
    /// Instead it will either start the daemon process and exit, or panic
    /// if daemonization fails.
    pub fn run(mut self, run_args: RunArgs) {
        info!("Server starting");

        let conf = self.configuration.as_ref();

        #[cfg(unix)]
        if conf.daemon {
            info!("Daemonizing the server");
            fast_timeout::pause_for_fork();
            let daemonize_result = daemonize(&self.configuration);
            fast_timeout::unpause();
            // If daemon_wait_for_ready is enabled, pass the parent PID to bootstrap so it
            // can send SIGUSR1 to the parent after bootstrap completes.
            if let Some(pid) = daemonize_result.notify_parent_pid {
                self.bootstrap.lock().set_notify_parent_pid(pid);
            }
        }

        #[cfg(windows)]
        if conf.daemon {
            panic!("Daemonizing under windows is not supported");
        }

        let blocking_opts = BlockingPoolOpts {
            max_threads: conf.max_blocking_threads,
            thread_keep_alive: conf.blocking_threads_ttl_seconds.map(Duration::from_secs),
        };
        let runtime_opts = conf.runtime_opts();
        if conf.runtime_enable_alt_timer && !conf.work_stealing {
            warn!("runtime_enable_alt_timer is ignored when work_stealing is disabled");
        }
        // This global timeout threshold is intended to be configured once during server startup,
        // before service runtimes begin creating timeout futures.
        fast_timeout::set_fast_timeout_to_tokio_threshold(
            conf.fast_timeout_to_tokio_threshold_seconds
                .map(Duration::from_secs),
        );

        // Initialize (or re-initialize) sentry and persist the guard for
        // the lifetime of the server. When daemonizing, the transport
        // thread spawned by any earlier `sentry::init` during
        // `bootstrap()` is lost after `fork()`, so a fresh init in the
        // child process is required. In non-daemon mode this is the
        // authoritative initialization that keeps sentry active.
        #[cfg(feature = "sentry")]
        self.bootstrap.lock().start_sentry();

        // Holds tuples of runtimes and their service name.
        let mut runtimes: Vec<(Runtime, String)> = Vec::new();

        // Set this before the bootstrap service loads inherited fds.
        #[cfg(unix)]
        if let Some(expected) = self.collect_listen_addresses() {
            self.bootstrap.lock().set_expected_listen_addrs(expected);
        }

        // Get services in topological order (dependencies first)
        let startup_order = match self.dependencies.lock().topological_sort() {
            Ok(order) => order,
            Err(e) => {
                error!("Failed to determine service startup order: {}", e);
                std::process::exit(1);
            }
        };

        // Log service names in startup order
        let service_names: Vec<String> = startup_order
            .iter()
            .map(|(_, service)| service.name.clone())
            .collect();
        info!("Starting services in dependency order: {:?}", service_names);

        // Start services in dependency order
        for (service_id, service) in startup_order {
            let mut wrapper = match self.services.remove(&service_id) {
                Some(w) => w,
                None => {
                    warn!(
                        "Service ID {:?}-{} in startup order but not found",
                        service_id, service.name
                    );
                    continue;
                }
            };

            let threads = wrapper.service.threads().unwrap_or(conf.threads);
            let name = wrapper.service.name().to_string();
            let service_runtime_opts = wrapper
                .service
                .runtime_opts_override(&runtime_opts)
                .unwrap_or_else(|| runtime_opts.clone());

            // Extract dependency watches from the ServiceHandle
            let dependencies = self
                .dependencies
                .lock()
                .get_dependencies(wrapper.service_handle.id);

            // Get the readiness notifier for this service by taking it from the Option.
            // Since service_id is the index, we can directly access it.
            // We take() the notifier, leaving None in its place.
            let ready_notifier = wrapper
                .ready_notifier
                .take()
                .expect("Service notifier should exist");

            if !dependencies.is_empty() {
                info!(
                    "Service '{name}' will wait for dependencies: {:?}",
                    dependencies.iter().map(|s| &s.name).collect::<Vec<_>>()
                );
            } else {
                info!("Starting service: {}", name);
            }

            let dependency_watches = dependencies
                .iter()
                .map(|s| s.ready_watch.clone())
                .collect::<Vec<_>>();

            let runtime = Server::run_service(
                wrapper.service,
                #[cfg(unix)]
                self.listen_fds(),
                self.shutdown_recv.clone(),
                threads,
                conf.work_stealing,
                self.configuration.listener_tasks_per_fd,
                ready_notifier,
                dependency_watches,
                blocking_opts.clone(),
                service_runtime_opts,
            );
            runtimes.push((runtime, name));
        }

        // blocked on main loop so that it runs forever
        // Only work steal runtime can use block_on()
        let server_runtime = Server::create_runtime(
            "Server",
            1,
            true,
            BlockingPoolOpts::default(),
            RuntimeOpts::default(),
        );
        #[cfg(unix)]
        let shutdown_type = server_runtime
            .get_handle()
            .block_on(self.main_loop(run_args));
        #[cfg(windows)]
        let shutdown_type = server_runtime
            .get_handle()
            .block_on(self.main_loop(run_args));

        self.execution_phase_watch
            .send(ExecutionPhase::ShutdownStarted)
            .ok();

        if matches!(shutdown_type, ShutdownType::Graceful) {
            self.execution_phase_watch
                .send(ExecutionPhase::ShutdownGracePeriod)
                .ok();

            let exit_timeout = self
                .configuration
                .as_ref()
                .grace_period_seconds
                .unwrap_or(EXIT_TIMEOUT);
            info!("Graceful shutdown: grace period {}s starts", exit_timeout);
            let grace_period = Duration::from_secs(exit_timeout);
            if wait_for_graceful_shutdown_idle(self.graceful_shutdown_idle.as_ref(), grace_period) {
                info!("Graceful shutdown: accepted work became idle");
            } else {
                info!("Graceful shutdown: grace period expired");
            }
            info!("Graceful shutdown: grace period ends");
        }

        // Give tokio runtimes time to exit
        let shutdown_timeout = match shutdown_type {
            ShutdownType::Quick => Duration::from_secs(0),
            ShutdownType::Graceful => Duration::from_secs(
                self.configuration
                    .as_ref()
                    .graceful_shutdown_timeout_seconds
                    .unwrap_or(5),
            ),
        };

        self.execution_phase_watch
            .send(ExecutionPhase::ShutdownRuntimes)
            .ok();

        let shutdowns: Vec<_> = runtimes
            .into_iter()
            .map(|(rt, name)| {
                info!("Waiting for runtimes to exit!");
                let join = thread::spawn(move || {
                    let start = Instant::now();
                    rt.shutdown_timeout(shutdown_timeout);
                    start.elapsed()
                });
                (join, name)
            })
            .collect();
        for (shutdown, name) in shutdowns {
            info!("Waiting for service runtime {name} to exit");
            match shutdown.join() {
                Ok(elapsed) if !shutdown_timeout.is_zero() && elapsed >= shutdown_timeout => {
                    warn!("Service runtime {name} did not exit within {shutdown_timeout:?}")
                }
                Ok(elapsed) => info!("Service runtime {name} exited after {elapsed:?}"),
                Err(e) => error!("Failed to shutdown service runtime {name}: {e:?}"),
            }
        }
        info!("All runtimes exited, exiting now");

        self.execution_phase_watch
            .send(ExecutionPhase::Terminated)
            .ok();
    }

    fn create_runtime(
        name: &str,
        threads: usize,
        work_steal: bool,
        blocking_opts: BlockingPoolOpts,
        runtime_opts: RuntimeOpts,
    ) -> Runtime {
        RuntimeBuilder::new(threads, name)
            .work_steal(work_steal)
            .blocking_pool_opts(blocking_opts)
            .runtime_opts(runtime_opts)
            .build()
    }
}

#[cfg(all(test, unix))]
mod listen_address_tests {
    use super::*;

    struct UnknownService;

    #[async_trait]
    impl crate::services::Service for UnknownService {
        fn name(&self) -> &str {
            "unknown"
        }
    }

    #[test]
    fn unknown_service_disables_inherited_fd_cleanup() {
        let mut server = Server::new_with_opt_and_conf(None, ServerConf::default());
        server.bootstrap_as_a_service();
        assert_eq!(server.collect_listen_addresses(), Some(Default::default()));

        server.add_service(UnknownService);
        assert_eq!(server.collect_listen_addresses(), None);
    }
}

#[cfg(test)]
mod graceful_shutdown_tests {
    use super::*;
    use std::sync::atomic::{AtomicBool, Ordering};

    #[test]
    fn idle_callback_short_circuits_grace_period() {
        let called = Arc::new(AtomicBool::new(false));
        let called_by_waiter = Arc::clone(&called);
        let idle_wait: GracefulShutdownIdle = Arc::new(move |grace_period| {
            assert_eq!(grace_period, Duration::from_secs(30));
            called_by_waiter.store(true, Ordering::Relaxed);
            true
        });

        assert!(
            wait_for_graceful_shutdown_idle(Some(&idle_wait), Duration::from_secs(30)),
            "the idle callback should report that work is complete"
        );
        assert!(
            called.load(Ordering::Relaxed),
            "the idle callback was not invoked"
        );
    }
}