pingora-core 0.9.1

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

#[cfg(unix)]
pub use super::transfer_fd::Fds;
use async_trait::async_trait;
#[cfg(unix)]
use log::warn;
use log::{debug, error, info};
use parking_lot::Mutex;
#[cfg(unix)]
use std::collections::HashSet;
use std::sync::Arc;
use tokio::sync::broadcast;

#[cfg(feature = "sentry")]
use sentry::ClientOptions;

#[cfg(unix)]
use crate::server::daemon::notify_parent_ready_for_fds;
#[cfg(unix)]
use crate::server::ListenFds;
#[cfg(unix)]
use std::time::Duration;

use crate::{
    prelude::Opt,
    server::{configuration::ServerConf, ExecutionPhase, ShutdownWatch},
    services::{background::BackgroundService, ServiceReadyNotifier},
};

/// Default timeout for retrying `SIGUSR1` to the parent when it fails with `EPERM`.
#[cfg(unix)]
const DEFAULT_DAEMON_NOTIFY_TIMEOUT: Duration = Duration::from_secs(60);

/// Service that allows the bootstrap process to be delayed until after
/// dependencies are ready
pub struct BootstrapService {
    inner: Arc<Mutex<Bootstrap>>,
}

impl BootstrapService {
    pub fn new(inner: &Arc<Mutex<Bootstrap>>) -> Self {
        BootstrapService {
            inner: Arc::clone(inner),
        }
    }
}

/// Encapsulation of the data needed to bootstrap the server
pub struct Bootstrap {
    completed: bool,

    test: bool,
    upgrade: bool,

    upgrade_sock: String,

    execution_phase_watch: broadcast::Sender<ExecutionPhase>,

    #[cfg(unix)]
    listen_fds: ListenFds,

    #[cfg(unix)]
    expected_listen_addrs: Option<HashSet<String>>,

    /// PID of the original parent process to notify via `SIGUSR1` after bootstrap completes.
    /// Set when [`ServerConf::daemon_wait_for_ready`] is `true`.
    #[cfg(unix)]
    notify_parent_pid: Option<u32>,

    /// How long to keep retrying `SIGUSR1` to the parent when it fails with `EPERM`.
    /// See [`ServerConf::daemon_notify_timeout_seconds`].
    #[cfg(unix)]
    daemon_notify_timeout: std::time::Duration,

    #[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 sentry: Option<ClientOptions>,

    /// The Sentry [`ClientInitGuard`](sentry::ClientInitGuard) returned by
    /// [`sentry::init`].
    ///
    /// This guard must be kept alive for the lifetime of the server, because
    /// dropping it flushes and disables the Sentry client.
    #[cfg(all(not(debug_assertions), feature = "sentry"))]
    sentry_guard: Option<sentry::ClientInitGuard>,
}

impl Bootstrap {
    pub fn new(
        options: &Option<Opt>,
        conf: &ServerConf,
        execution_phase_watch: &broadcast::Sender<ExecutionPhase>,
    ) -> Self {
        let (test, upgrade) = options
            .as_ref()
            .map(|opt| (opt.test, opt.upgrade))
            .unwrap_or_default();

        let upgrade_sock = conf.upgrade_sock.clone();

        Bootstrap {
            test,
            upgrade,
            upgrade_sock,
            #[cfg(unix)]
            listen_fds: Arc::new(Mutex::new(Fds::new())),
            #[cfg(unix)]
            expected_listen_addrs: None,
            #[cfg(unix)]
            notify_parent_pid: None,
            #[cfg(unix)]
            daemon_notify_timeout: conf
                .daemon_notify_timeout_seconds
                .map(|n| Duration::from_secs(n.get()))
                .unwrap_or(DEFAULT_DAEMON_NOTIFY_TIMEOUT),
            execution_phase_watch: execution_phase_watch.clone(),
            completed: false,
            #[cfg(feature = "sentry")]
            sentry: None,
            #[cfg(all(not(debug_assertions), feature = "sentry"))]
            sentry_guard: None,
        }
    }

    #[cfg(feature = "sentry")]
    pub fn set_sentry_config(&mut self, sentry_config: Option<ClientOptions>) {
        self.sentry = sentry_config;
    }

    /// Store the parent process PID to notify via `SIGUSR1` after bootstrap completes.
    /// Only relevant when [`ServerConf::daemon_wait_for_ready`] is `true`.
    #[cfg(unix)]
    pub fn set_notify_parent_pid(&mut self, pid: u32) {
        self.notify_parent_pid = Some(pid);
    }

    /// Set the expected bind addresses and prune any inherited sockets already loaded.
    #[cfg(unix)]
    pub fn set_expected_listen_addrs(&mut self, addrs: HashSet<String>) {
        Self::close_unclaimed_fds(&mut self.listen_fds.lock(), &addrs);
        self.expected_listen_addrs = Some(addrs);
    }

    /// Initialize the Sentry client from the configured [`ClientOptions`] and
    /// store the resulting guard.
    ///
    /// The [`ClientOptions`] are preserved (not consumed) so that sentry can be
    /// re-initialized after daemonization, when the transport thread spawned by
    /// the previous [`sentry::init`] call is lost due to `fork()`.
    ///
    /// The resulting [`sentry::ClientInitGuard`] is stored in `self` so that it
    /// lives as long as the [`Bootstrap`] (and therefore the
    /// [`Server`](super::Server)), keeping the Sentry client active for the
    /// lifetime of the process.
    ///
    /// Sentry is only initialized in release builds; in debug builds this is a
    /// no-op.
    #[cfg(feature = "sentry")]
    pub(super) fn start_sentry(&mut self) {
        #[cfg(not(debug_assertions))]
        {
            self.sentry_guard = self.sentry.as_ref().map(|opts| sentry::init(opts.clone()));
        }
    }

    pub fn bootstrap(&mut self) {
        // already bootstrapped
        if self.completed {
            return;
        }

        info!("Bootstrap starting");

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

        // Temporarily initialize sentry if it isn't already active, so that
        // errors during fd loading are captured. If sentry was already
        // initialized with a persistent guard by `Server::run()` (as in
        // the `bootstrap_as_a_service` path), we skip this to avoid
        // clobbering the persistent guard.
        #[cfg(all(not(debug_assertions), feature = "sentry"))]
        let _guard = if self.sentry_guard.is_none() {
            self.sentry.as_ref().map(|opts| sentry::init(opts.clone()))
        } else {
            None
        };

        if self.test {
            info!("Server Test passed, exiting");
            std::process::exit(0);
        }

        // Notify the parent process that it can exit. It might seem like we should load the file
        // descriptors from the old process first, but the purpose of this notification is to
        // release the parent so that the process managing it (e.g. systemd) can continue and send
        // a quit signal to the old process. That quit signal is required before the old process
        // will start trying to send its file descriptors to us — so if we called load_fds first,
        // we would be guaranteeing a timeout.
        #[cfg(unix)]
        if let Some(pid) = self.notify_parent_pid {
            notify_parent_ready_for_fds(pid, self.daemon_notify_timeout);
        }

        #[cfg(unix)]
        match self.load_fds(self.upgrade) {
            Ok(_) => {
                info!("Bootstrap done");
            }
            Err(e) => {
                // sentry log error on fd load failure
                #[cfg(all(not(debug_assertions), feature = "sentry"))]
                sentry::capture_error(&e);

                error!("Bootstrap failed on error: {:?}, exiting.", e);
                std::process::exit(1);
            }
        }

        self.completed = true;

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

    #[cfg(unix)]
    fn load_fds(&mut self, upgrade: bool) -> Result<(), nix::Error> {
        if upgrade {
            debug!("Trying to receive socks");
            let mut fds = Fds::new();
            fds.get_from_sock(self.upgrade_sock.as_str())?;
            if let Some(expected) = &self.expected_listen_addrs {
                Self::close_unclaimed_fds(&mut fds, expected);
            }
            // Mutate through the existing Arc so all clones held by services see the update.
            *self.listen_fds.lock() = fds;
        }
        Ok(())
    }

    #[cfg(unix)]
    fn close_unclaimed_fds(fds: &mut Fds, expected: &HashSet<String>) {
        let closed = fds.close_unclaimed(expected);
        if !closed.is_empty() {
            warn!(
                "Closed {} inherited listening socket(s) not claimed by any service: {:?}",
                closed.len(),
                closed
            );
        }
    }

    #[cfg(unix)]
    pub fn get_fds(&self) -> ListenFds {
        self.listen_fds.clone()
    }
}

#[async_trait]
impl BackgroundService for BootstrapService {
    async fn start_with_ready_notifier(
        &self,
        _shutdown: ShutdownWatch,
        notifier: ServiceReadyNotifier,
    ) {
        self.inner.lock().bootstrap();
        notifier.notify_ready();
    }
}

#[cfg(all(test, unix))]
mod tests {
    use super::*;
    use std::io::{ErrorKind, Read};
    use std::os::unix::io::{FromRawFd, IntoRawFd, OwnedFd};
    use std::os::unix::net::UnixStream;

    #[test]
    fn expected_addresses_prune_already_loaded_fds() {
        let (execution_phase_watch, _) = broadcast::channel(1);
        let mut bootstrap = Bootstrap::new(&None, &ServerConf::default(), &execution_phase_watch);

        // Closure is observed by reading the other end of a socket pair, which
        // reports end-of-file exactly when its peer is gone. Checking the
        // descriptor *number* would be racy: numbers are process-wide and
        // reassigned lowest-free-first, so a concurrent test in this binary can
        // be handed the number just closed and make it valid again first.
        // Observing ends are non-blocking, so a descriptor that wrongly stayed
        // open fails an assertion rather than hanging.
        //
        // A second owning handle would turn a wrongly closed descriptor into an
        // I/O safety abort of the whole binary, so `Fds` gets raw numbers.
        let (mut keep_peer, keep_local) = UnixStream::pair().unwrap();
        let (mut drop_peer, drop_local) = UnixStream::pair().unwrap();
        keep_peer.set_nonblocking(true).unwrap();
        drop_peer.set_nonblocking(true).unwrap();
        let keep_fd = keep_local.into_raw_fd();

        {
            let mut fds = bootstrap.listen_fds.lock();
            fds.add("127.0.0.1:80".to_string(), keep_fd);
            fds.add("127.0.0.1:9090".to_string(), drop_local.into_raw_fd());
        }

        bootstrap.set_expected_listen_addrs(["127.0.0.1:80".to_string()].into_iter().collect());

        let fds = bootstrap.listen_fds.lock();
        assert_eq!(*fds.get("127.0.0.1:80").unwrap(), keep_fd);
        assert!(fds.get("127.0.0.1:9090").is_none());
        drop(fds);
        assert!(
            matches!(drop_peer.read(&mut [0u8; 1]), Ok(0)),
            "the pruned descriptor was not closed"
        );
        assert!(
            matches!(keep_peer.read(&mut [0u8; 1]), Err(e) if e.kind() == ErrorKind::WouldBlock),
            "the expected descriptor was closed"
        );

        // `Fds` does not close on drop.
        // SAFETY: asserted open just above, and no other owner remains.
        drop(unsafe { OwnedFd::from_raw_fd(keep_fd) });
    }
}