ironflow_ops_postgres/admin/
process.rs

1//! Backend process management operations.
2
3use async_trait::async_trait;
4use ironflow_core::error::OperationError;
5use ironflow_core::operation::{Operation, OperationContext, TypedOperation};
6use serde::{Deserialize, Serialize};
7use serde_json::Value;
8use sqlx::{PgPool, Row};
9
10use crate::helpers::{pg_error, to_value};
11
12/// Output of [`CancelQuery`].
13#[derive(Debug, Clone, Serialize, Deserialize)]
14pub struct CancelQueryOutput {
15    /// Whether the cancellation signal was sent successfully.
16    pub cancelled: bool,
17}
18
19/// Cancel a running query by its backend process ID.
20///
21/// # Examples
22///
23/// ```no_run
24/// use ironflow_ops_postgres::admin::process::CancelQuery;
25/// use ironflow_core::operation::Operation;
26///
27/// # fn example(pool: sqlx::PgPool) {
28/// let op = CancelQuery::new(pool, 12345);
29/// assert_eq!(op.kind(), "postgres");
30/// # }
31/// ```
32pub struct CancelQuery {
33    pool: PgPool,
34    pid: i32,
35}
36
37impl CancelQuery {
38    /// Create a new cancel-query operation.
39    pub fn new(pool: PgPool, pid: i32) -> Self {
40        Self { pool, pid }
41    }
42
43    /// Execute and return a typed result.
44    ///
45    /// # Errors
46    ///
47    /// Returns [`OperationError::External`] on connection errors.
48    pub async fn run(&self, _ctx: &OperationContext) -> Result<CancelQueryOutput, OperationError> {
49        let row = sqlx::query("SELECT pg_cancel_backend($1) AS cancelled")
50            .bind(self.pid)
51            .fetch_one(&self.pool)
52            .await
53            .map_err(pg_error)?;
54        let cancelled: bool = row.try_get("cancelled").map_err(pg_error)?;
55        Ok(CancelQueryOutput { cancelled })
56    }
57}
58
59#[async_trait]
60impl Operation for CancelQuery {
61    fn kind(&self) -> &str {
62        "postgres"
63    }
64    async fn execute(&self, ctx: &OperationContext) -> Result<Value, OperationError> {
65        to_value(&self.run(ctx).await?)
66    }
67    fn input(&self) -> Option<Value> {
68        Some(serde_json::json!({ "pid": self.pid }))
69    }
70}
71
72impl TypedOperation for CancelQuery {
73    type Output = CancelQueryOutput;
74}
75
76/// Output of [`TerminateBackend`].
77#[derive(Debug, Clone, Serialize, Deserialize)]
78pub struct TerminateBackendOutput {
79    /// Whether the backend was terminated.
80    pub terminated: bool,
81}
82
83/// Terminate a backend process by its PID.
84///
85/// This is more forceful than [`CancelQuery`] and should be used as a last
86/// resort.
87///
88/// # Examples
89///
90/// ```no_run
91/// use ironflow_ops_postgres::admin::process::TerminateBackend;
92/// use ironflow_core::operation::Operation;
93///
94/// # fn example(pool: sqlx::PgPool) {
95/// let op = TerminateBackend::new(pool, 12345);
96/// assert_eq!(op.kind(), "postgres");
97/// # }
98/// ```
99pub struct TerminateBackend {
100    pool: PgPool,
101    pid: i32,
102}
103
104impl TerminateBackend {
105    /// Create a new terminate-backend operation.
106    pub fn new(pool: PgPool, pid: i32) -> Self {
107        Self { pool, pid }
108    }
109
110    /// Execute and return a typed result.
111    ///
112    /// # Errors
113    ///
114    /// Returns [`OperationError::External`] on connection errors.
115    pub async fn run(
116        &self,
117        _ctx: &OperationContext,
118    ) -> Result<TerminateBackendOutput, OperationError> {
119        let row = sqlx::query("SELECT pg_terminate_backend($1) AS terminated")
120            .bind(self.pid)
121            .fetch_one(&self.pool)
122            .await
123            .map_err(pg_error)?;
124        let terminated: bool = row.try_get("terminated").map_err(pg_error)?;
125        Ok(TerminateBackendOutput { terminated })
126    }
127}
128
129#[async_trait]
130impl Operation for TerminateBackend {
131    fn kind(&self) -> &str {
132        "postgres"
133    }
134    async fn execute(&self, ctx: &OperationContext) -> Result<Value, OperationError> {
135        to_value(&self.run(ctx).await?)
136    }
137    fn input(&self) -> Option<Value> {
138        Some(serde_json::json!({ "pid": self.pid }))
139    }
140}
141
142impl TypedOperation for TerminateBackend {
143    type Output = TerminateBackendOutput;
144}
145
146#[cfg(test)]
147mod tests {
148    use super::*;
149
150    #[tokio::test]
151    async fn cancel_query_kind() {
152        let pool = PgPool::connect_lazy("postgres://localhost/test").unwrap();
153        let op = CancelQuery::new(pool, 123);
154        assert_eq!(op.kind(), "postgres");
155    }
156
157    #[tokio::test]
158    async fn terminate_backend_kind() {
159        let pool = PgPool::connect_lazy("postgres://localhost/test").unwrap();
160        let op = TerminateBackend::new(pool, 123);
161        assert_eq!(op.kind(), "postgres");
162    }
163
164    #[tokio::test]
165    async fn cancel_query_input_no_secrets() {
166        let pool = PgPool::connect_lazy("postgres://localhost/test").unwrap();
167        let op = CancelQuery::new(pool, 123);
168        let input = op.input().unwrap();
169        let text = input.to_string();
170        assert!(!text.contains("postgres://"), "leaked URL: {text}");
171    }
172}
ironflow_ops_postgres/admin/process.rs

ironflow_ops_postgres/admin/
process.rs
