feat(procmgr): reject RPCs during reload and shutdown

Add an OperationGate so config reload and daemon shutdown block list,
describe, get_status, get_config, create, start, and stop with
FAILED_PRECONDITION while in progress.

Author: jose-manuel-almaza

pkg/procmgr/rust/src/grpc/service.rs

@@ -31,6 +31,10 @@ impl ProcessManagerService {
             cmd_tx,
         }
     }
+
+    async fn ensure_idle(&self) -> Result<(), Status> {
+        self.mgr.ensure_idle().await
+    }
 }
 
 #[tonic::async_trait]
@@ -39,6 +43,7 @@ impl proto::process_manager_server::ProcessManager for ProcessManagerService {
         &self,
         _request: Request<proto::ListRequest>,
     ) -> Result<Response<proto::ListResponse>, Status> {
+        self.ensure_idle().await?;
         let procs = self.mgr.processes().await;
         let processes = procs.iter().map(process_to_proto).collect();
         Ok(Response::new(proto::ListResponse { processes }))
@@ -48,6 +53,7 @@ impl proto::process_manager_server::ProcessManager for ProcessManagerService {
         &self,
         request: Request<proto::DescribeRequest>,
     ) -> Result<Response<proto::DescribeResponse>, Status> {
+        self.ensure_idle().await?;
         let name_or_uuid = request.into_inner().name_or_uuid;
         let (mut detail, pid) = {
             let procs = self.mgr.processes().await;
@@ -67,6 +73,7 @@ impl proto::process_manager_server::ProcessManager for ProcessManagerService {
         &self,
         _request: Request<proto::GetStatusRequest>,
     ) -> Result<Response<proto::GetStatusResponse>, Status> {
+        self.ensure_idle().await?;
         let procs = self.mgr.processes().await;
         let total = procs.len() as u32;
         let (mut created, mut starting, mut running, mut stopping) = (0u32, 0, 0, 0);
@@ -103,6 +110,7 @@ impl proto::process_manager_server::ProcessManager for ProcessManagerService {
         request: Request<proto::CreateRequest>,
     ) -> Result<Response<proto::CreateResponse>, Status> {
         require_privileged_pipe_client(&request)?;
+        self.ensure_idle().await?;
         let req = request.into_inner();
         let config = create_request_to_config(&req)?;
         let (reply_tx, reply_rx) = oneshot::channel();
@@ -129,6 +137,7 @@ impl proto::process_manager_server::ProcessManager for ProcessManagerService {
         &self,
         request: Request<proto::StartRequest>,
     ) -> Result<Response<proto::StartResponse>, Status> {
+        self.ensure_idle().await?;
         let name_or_uuid = request.into_inner().name_or_uuid;
         let (reply_tx, reply_rx) = oneshot::channel();
         self.cmd_tx
@@ -154,6 +163,7 @@ impl proto::process_manager_server::ProcessManager for ProcessManagerService {
         &self,
         request: Request<proto::StopRequest>,
     ) -> Result<Response<proto::StopResponse>, Status> {
+        self.ensure_idle().await?;
         let name_or_uuid = request.into_inner().name_or_uuid;
         let (reply_tx, reply_rx) = oneshot::channel();
         self.cmd_tx
@@ -198,6 +208,7 @@ impl proto::process_manager_server::ProcessManager for ProcessManagerService {
         &self,
         _request: Request<proto::GetConfigRequest>,
     ) -> Result<Response<proto::GetConfigResponse>, Status> {
+        self.ensure_idle().await?;
         let procs = self.mgr.processes().await;
         let runtime = procs
             .iter()

pkg/procmgr/rust/src/lib.rs

@@ -9,6 +9,7 @@ pub mod env;
 pub mod grpc;
 pub mod handle;
 pub mod manager;
+mod operation;
 pub mod ordering;
 pub mod platform;
 pub mod process;

pkg/procmgr/rust/src/manager/process_manager.rs

@@ -4,6 +4,7 @@ use super::{
 };
 use crate::command::{CreateResult, StartResult, StopResult};
 use crate::config::{self, ConfigLoader, ProcessDefinition};
+use crate::operation::{OperationGate, OperationKind};
 use crate::ordering;
 use crate::process::ManagedProcess;
 use crate::shutdown;
@@ -20,6 +21,7 @@ pub struct ProcessManager {
     pub(in crate::manager) startup_order: Arc<RwLock<Vec<usize>>>,
     pub(in crate::manager) config_loader: Arc<dyn ConfigLoader>,
     pub(in crate::manager) uuid_gen: Arc<dyn UuidGenerator>,
+    pub(in crate::manager) operation_gate: OperationGate,
 }
 
 impl ProcessManager {
@@ -38,9 +40,20 @@ impl ProcessManager {
             startup_order: Arc::new(RwLock::new(startup_result.order)),
             config_loader,
             uuid_gen,
+            operation_gate: OperationGate::default(),
         }
     }
 
+    pub(crate) async fn ensure_idle(&self) -> Result<(), Status> {
+        self.operation_gate.ensure_idle().await
+    }
+
+    pub(in crate::manager) async fn force_begin_shutdown(&self) {
+        self.operation_gate
+            .force_begin(OperationKind::Shutdown)
+            .await;
+    }
+
     /// Wrap this manager in a [`Supervisor`] for daemon execution.
     pub fn supervisor(self) -> Supervisor {
         Supervisor::new(self)

pkg/procmgr/rust/src/manager/reload.rs

@@ -2,6 +2,7 @@ use super::supervisor::RuntimeHandles;
 use super::{ProcessManager, queue_restart, try_spawn_and_watch};
 use crate::command::ReloadResult;
 use crate::config::ProcessDefinition;
+use crate::operation::OperationKind;
 use crate::process::{ManagedProcess, ProcessOrigin};
 use crate::state::ProcessState;
 use log::{info, warn};
@@ -122,6 +123,20 @@ impl ProcessManager {
     pub(crate) async fn handle_reload_config(
         &self,
         handles: &RuntimeHandles,
+    ) -> Result<ReloadResult, Status> {
+        self.operation_gate
+            .try_begin(OperationKind::ReloadConfig)
+            .await?;
+        let result = self.handle_reload_config_impl(handles).await;
+        self.operation_gate
+            .end(OperationKind::ReloadConfig)
+            .await;
+        result
+    }
+
+    async fn handle_reload_config_impl(
+        &self,
+        handles: &RuntimeHandles,
     ) -> Result<ReloadResult, Status> {
         let new_configs = self.config_loader.load();
 

pkg/procmgr/rust/src/manager/supervisor.rs

@@ -118,7 +118,7 @@ impl Supervisor {
 
         let shutdown = platform::shutdown_signal();
         tokio::pin!(shutdown);
-        run_manager_event_loop(
+        let shutdown_requested = run_manager_event_loop(
             &manager,
             &handles,
             &mut cmd_rx,
@@ -128,6 +128,10 @@ impl Supervisor {
         )
         .await;
 
+        if shutdown_requested {
+            manager.force_begin_shutdown().await;
+        }
+
         info!("dd-procmgrd shutting down");
 
         let _ = grpc_shutdown_tx.send(());

pkg/procmgr/rust/src/operation.rs

@@ -0,0 +1,82 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2026-present Datadog, Inc.
+
+use std::sync::Arc;
+
+use tokio::sync::Mutex;
+use tonic::Status;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum OperationKind {
+    ReloadConfig,
+    Shutdown,
+}
+
+#[derive(Clone, Default)]
+pub(crate) struct OperationGate {
+    active: Arc<Mutex<Option<OperationKind>>>,
+}
+
+impl OperationGate {
+    pub async fn try_begin(&self, op: OperationKind) -> Result<(), Status> {
+        let mut guard = self.active.lock().await;
+        if let Some(active) = *guard {
+            return Err(Status::failed_precondition(format!(
+                "operation in progress ({active:?}); try again later"
+            )));
+        }
+        *guard = Some(op);
+        Ok(())
+    }
+
+    pub async fn force_begin(&self, op: OperationKind) {
+        *self.active.lock().await = Some(op);
+    }
+
+    pub async fn end(&self, op: OperationKind) {
+        let mut guard = self.active.lock().await;
+        if *guard == Some(op) {
+            *guard = None;
+        }
+    }
+
+    pub async fn ensure_idle(&self) -> Result<(), Status> {
+        let guard = self.active.lock().await;
+        if let Some(active) = *guard {
+            return Err(Status::failed_precondition(format!(
+                "operation in progress ({active:?}); try again later"
+            )));
+        }
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn test_operation_gate_blocks_while_active() {
+        let gate = OperationGate::default();
+        gate.try_begin(OperationKind::ReloadConfig)
+            .await
+            .expect("first begin");
+        assert!(gate.ensure_idle().await.is_err());
+        assert!(gate.try_begin(OperationKind::ReloadConfig).await.is_err());
+        gate.end(OperationKind::ReloadConfig).await;
+        gate.ensure_idle().await.expect("idle after end");
+    }
+
+    #[tokio::test]
+    async fn test_force_begin_shutdown_replaces_active_operation() {
+        let gate = OperationGate::default();
+        gate.try_begin(OperationKind::ReloadConfig)
+            .await
+            .expect("reload begin");
+        gate.force_begin(OperationKind::Shutdown).await;
+        assert!(gate.ensure_idle().await.is_err());
+        assert!(gate.try_begin(OperationKind::ReloadConfig).await.is_err());
+    }
+}