Allow opted-in services to keep accepting during graceful shutdown

.github/workflows/build.yml

@@ -96,6 +96,15 @@ jobs:
           cargo nextest run --workspace --no-fail-fast
           -E 'not test(test_h2_late_frames_on_cancelled_streams_keep_connection)'
 
+      - name: Test keep listening on terminate
+        if: matrix.toolchain == '1.97.1'
+        run: |
+          cargo clippy --package pingora-core --all-targets --features keep_listening_on_terminate -- --allow=unknown-lints --deny=warnings
+          cargo nextest run --package pingora-core --features keep_listening_on_terminate \
+            --lib --test test_keep_listening_on_terminate --no-fail-fast \
+            -E 'not test(test_h2_late_frames_on_cancelled_streams_keep_connection)'
+          cargo build --package pingora-proxy --example keep_listening_on_terminate --features keep_listening_on_terminate
+
       # nextest does not run doc tests, so run them separately.
       # (https://github.com/rust-lang/cargo/issues/6669)
       - name: Run cargo doc test

docs/user_guide/start_stop.md

@@ -23,5 +23,12 @@ Upon receiving SIGINT (ctrl + c), the server will exit immediately with no delay
 ### SIGTERM: graceful shutdown
 Upon receiving SIGTERM, the server will notify all its services to shutdown, wait for some preconfigured time and then exit. This behavior gives requests a grace period to finish.
 
+Listening services stop accepting new connections when shutdown starts. A service that must stay reachable during the grace period, such as a readiness or metrics endpoint, can opt in to keep accepting with `Service::keep_listening_on_terminate()`, behind the `keep_listening_on_terminate` cargo feature ([example](../../pingora-proxy/examples/keep_listening_on_terminate.rs)). The application still reports draining itself, for example from a background service's `ShutdownWatch`.
+
+| Signal | Default listener | `keep_listening_on_terminate` |
+| ------ | ---------------- | ----------------------------- |
+| SIGTERM | Stops accepting | Accepts until the grace period ends |
+| SIGQUIT | Transfers its socket, then stops accepting | Same as default |
+
 ### SIGQUIT: graceful upgrade
 Similar to SIGTERM, but the server will also transfer all its listening sockets to a new Pingora server so that there is no downtime during the upgrade. See the [graceful upgrade](graceful.md) section for more details.

pingora-core/Cargo.toml

@@ -128,3 +128,4 @@ openssl_derived = ["any_tls"]
 any_tls = ["dep:tokio-util"]
 sentry = ["dep:sentry"]
 connection_filter = []
+keep_listening_on_terminate = []

pingora-core/src/services/listening.rs

@@ -41,6 +41,13 @@ use std::fs::Permissions;
 use std::marker::PhantomData;
 use std::sync::Arc;
 
+#[cfg(feature = "keep_listening_on_terminate")]
+mod keep_listening;
+#[cfg(all(test, feature = "keep_listening_on_terminate"))]
+mod keep_listening_tests;
+#[cfg(feature = "keep_listening_on_terminate")]
+use keep_listening::TerminatePhases;
+
 /// Override the runtime options used to run a listening service.
 pub type RuntimeOptsOverride = Arc<dyn Fn(&RuntimeOpts) -> Option<RuntimeOpts> + Send + Sync>;
 
@@ -57,6 +64,9 @@ where
     pub threads: Option<usize>,
     runtime_opts_override: Option<RuntimeOptsOverride>,
     runtime_group: Option<String>,
+    /// Set by [`Self::keep_listening_on_terminate`].
+    #[cfg(feature = "keep_listening_on_terminate")]
+    terminate_phases: Option<Arc<TerminatePhases>>,
     #[cfg(feature = "connection_filter")]
     connection_filter: Arc<dyn ConnectionFilter>,
 }
@@ -75,6 +85,8 @@ where
             threads: None,
             runtime_opts_override: None,
             runtime_group: None,
+            #[cfg(feature = "keep_listening_on_terminate")]
+            terminate_phases: None,
             #[cfg(feature = "connection_filter")]
             connection_filter: Arc::new(AcceptAllFilter),
         }
@@ -94,6 +106,8 @@ where
             threads: None,
             runtime_opts_override: None,
             runtime_group: None,
+            #[cfg(feature = "keep_listening_on_terminate")]
+            terminate_phases: None,
             #[cfg(feature = "connection_filter")]
             connection_filter: Arc::new(AcceptAllFilter),
         }
@@ -244,6 +258,7 @@ where
         app_logic: Arc<A>,
         mut stack: TransportStack,
         mut shutdown: ShutdownWatch,
+        #[cfg(feature = "keep_listening_on_terminate")] phases: Option<Arc<TerminatePhases>>,
     ) {
         // the accept loop, until the system is shutting down
         loop {
@@ -256,6 +271,17 @@ where
                                 // happen in the initial read
                                 continue;
                             }
+                            #[cfg(feature = "keep_listening_on_terminate")]
+                            if phases
+                                .as_ref()
+                                .is_some_and(|phases| !phases.is_graceful_upgrade())
+                            {
+                                info!(
+                                    "Graceful terminate: {} keeps listening until runtime shutdown",
+                                    stack.as_str()
+                                );
+                                continue;
+                            }
                             info!("Shutting down {}", stack.as_str());
                             break;
                         }
@@ -347,6 +373,8 @@ where
             .take()
             .expect("can only start_service() once");
         let app_logic = Arc::new(app_logic);
+        #[cfg(feature = "keep_listening_on_terminate")]
+        let terminate_phases = self.terminate_phases.take();
 
         let mut handlers = Vec::new();
 
@@ -355,9 +383,18 @@ where
                 let shutdown = shutdown.clone();
                 let my_app_logic = app_logic.clone();
                 let endpoint = endpoint.clone();
+                #[cfg(feature = "keep_listening_on_terminate")]
+                let phases = terminate_phases.clone();
 
                 let jh = runtime.spawn(async move {
-                    Self::run_endpoint(my_app_logic, endpoint, shutdown).await;
+                    Self::run_endpoint(
+                        my_app_logic,
+                        endpoint,
+                        shutdown,
+                        #[cfg(feature = "keep_listening_on_terminate")]
+                        phases,
+                    )
+                    .await;
                 });
 
                 handlers.push(jh);
@@ -472,6 +509,8 @@ mod tests {
             Arc::new(HoldUntilByte),
             stack,
             shutdown,
+            #[cfg(feature = "keep_listening_on_terminate")]
+            None,
         ));
 
         let mut first = TcpStream::connect(addr).await.unwrap();
@@ -508,6 +547,8 @@ mod tests {
             Arc::new(HandOff(streams_tx)),
             stack,
             shutdown,
+            #[cfg(feature = "keep_listening_on_terminate")]
+            None,
         ));
 
         let _first = TcpStream::connect(addr).await.unwrap();

pingora-core/src/services/listening/keep_listening.rs

@@ -0,0 +1,87 @@
+// 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.
+
+use super::Service;
+use crate::protocols::http::custom::server::Session as CustomServerSession;
+use crate::server::{ExecutionPhase, Server};
+use parking_lot::Mutex;
+use std::sync::Arc;
+use tokio::sync::broadcast::{self, error::TryRecvError};
+
+impl<A, DS> Service<A, DS>
+where
+    DS: CustomServerSession,
+{
+    /// Keep accepting connections through a graceful terminate (SIGTERM) until `server` shuts its
+    /// runtimes down, instead of closing the listeners when shutdown starts.
+    ///
+    /// Meant for endpoints that must stay reachable while the server drains, such as readiness
+    /// or metrics. The listeners still close on a graceful upgrade (SIGQUIT), so the new process
+    /// takes over the transferred sockets. `server` must be the server this service is added to.
+    ///
+    /// Only accepting is kept. The application still owns its drain behavior:
+    /// - On terminate this bypasses the normal cleanup path: runtime shutdown drops the
+    ///   listeners, so [`ServerApp::cleanup`](crate::apps::ServerApp::cleanup) is not called, and
+    ///   an HTTP proxy's `Session::is_process_shutting_down()` stays `false`. Observe shutdown
+    ///   separately, e.g. through a background service's
+    ///   [`ShutdownWatch`](crate::server::ShutdownWatch).
+    /// - Keepalive, response status, and readiness reporting are unchanged.
+    ///
+    /// A service that serves regular traffic should not opt in: it would keep admitting work that
+    /// runtime shutdown then drops.
+    pub fn keep_listening_on_terminate(&mut self, server: &Server) {
+        self.keep_listening_with(server.watch_execution_phase());
+    }
+
+    pub(super) fn keep_listening_with(&mut self, phases: broadcast::Receiver<ExecutionPhase>) {
+        self.terminate_phases = Some(Arc::new(TerminatePhases(Mutex::new(Decision {
+            phases,
+            upgrade: None,
+        }))));
+    }
+}
+
+/// Shared by a service's endpoints so they all make the same shutdown decision.
+pub(super) struct TerminatePhases(Mutex<Decision>);
+
+struct Decision {
+    phases: broadcast::Receiver<ExecutionPhase>,
+    upgrade: Option<bool>,
+}
+
+impl TerminatePhases {
+    /// Whether the shutdown just broadcast is a graceful upgrade.
+    ///
+    /// The server sends the upgrade phases before it broadcasts shutdown, so they are already
+    /// queued. Lost phases or a closed channel count as an upgrade, the default close.
+    pub(super) fn is_graceful_upgrade(&self) -> bool {
+        let mut decision = self.0.lock();
+        if let Some(upgrade) = decision.upgrade {
+            return upgrade;
+        }
+        let upgrade = loop {
+            match decision.phases.try_recv() {
+                Ok(
+                    ExecutionPhase::GracefulUpgradeTransferringFds
+                    | ExecutionPhase::GracefulUpgradeCloseTimeout,
+                )
+                | Err(TryRecvError::Lagged(_) | TryRecvError::Closed) => break true,
+                Ok(_) => {}
+                Err(TryRecvError::Empty) => break false,
+            }
+        };
+        decision.upgrade = Some(upgrade);
+        upgrade
+    }
+}