diff --git a/integration/rust/tests/integration/admin_reload.rs b/integration/rust/tests/integration/admin_reload.rs new file mode 100644 index 000000000..5e7cc7f51 --- /dev/null +++ b/integration/rust/tests/integration/admin_reload.rs @@ -0,0 +1,111 @@ +use sqlx::Row; +use std::time::Duration; + +use tokio::time::{Instant, sleep}; + +use crate::setup::{admin_sqlx, connections_sqlx}; + +/// +/// Test the implementation of the command FORCE_RELOAD, which is a normal RELOAD + terminates all in-flight transactions. +#[tokio::test] +async fn admin_force_reload_test() { + let admin = admin_sqlx().await; + let connections = connections_sqlx().await; // [pgdog, pgdog_sharded] + + // Test with an idle transaction. + { + let mut transaction = connections.get(1).unwrap().begin().await.unwrap(); + + // Isn't strictly needed for the functionaltiy of the test; but why not? + sqlx::raw_sql("SELECT * FROM sharded") + .fetch_all(&mut *transaction) + .await + .unwrap(); + + // After we force reload, existing transactions (i.e. this one) are terminated. + sqlx::raw_sql("FORCE_RELOAD").execute(&admin).await.unwrap(); + + let err = sqlx::raw_sql("SELECT * FROM sharded") + .fetch_all(&mut *transaction) + .await + .err() + .unwrap(); + + // Identical to Postgres error pertaining to normal backend termination. + // + assert!( + err.as_database_error() + .unwrap() + .message() + .contains("terminating connection due to administrator command") + ); + + // Should be overwritten from the generic error code to the same one as Postgres. + assert_eq!(err.as_database_error().unwrap().code().unwrap(), "57P01"); + + // The transaction drops (allowing another connection in sqlx `Pool`) + } + + // Does it work with a new Pool? + let test = connections_sqlx().await; + let test = test.get(1).unwrap(); + sqlx::raw_sql("SELECT 1234").fetch_all(test).await.unwrap(); + + // Does it still with the old Pool? + let conn = connections.get(1).unwrap(); + sqlx::raw_sql("SELECT 1000").fetch_all(conn).await.unwrap(); + + // Additionally try when we're in an active query (as opposed to an idle transaction) + { + let query = async { + let connections = connections_sqlx().await; + let mut transaction = connections.get(1).unwrap().begin().await.unwrap(); + + let start_ts = Instant::now(); + + // Arbitrary amount of time; it should be cancelled long before; lets us be sure that + // FORCE_RELOAD is actually cancelling it mid-query + let err = sqlx::raw_sql("SELECT pg_sleep(5)") + .execute(&mut *transaction) + .await + .err() + .unwrap(); + + assert!( + err.as_database_error() + .unwrap() + .message() + .contains("terminating connection due to administrator command") + ); + + // We're sleeping for 5s. Waiting for 500ms on `reload` async block. + // So it should be <1s until we get the admin termination error. + assert!(start_ts.elapsed() < Duration::from_secs(1)); + }; + + let reload = async { + sleep(Duration::from_millis(500)).await; + + // After we force reload, existing transactions (i.e. this one) are terminated. + sqlx::raw_sql("FORCE_RELOAD").execute(&admin).await.unwrap(); + }; + + // Runs both the query and reload concurrently + // Returns after both have finished. + tokio::join!(query, reload); + + // The transaction was terminated (through PgDog), however, is Postgres still running it? + let rows = sqlx::raw_sql( + "SELECT * FROM pg_stat_activity WHERE state = 'active' AND query NOT LIKE '%pg_stat_activity%'", + ) + .fetch_all(conn) + .await + .unwrap(); + + // I did it this way to prevent flaky tests if the health-check were to run in parallel + // Usually there's no rows. + for row in rows { + assert!(!row.get::<&str, &str>("query").eq("SELECT pg_sleep(5)")); + } + } +} diff --git a/integration/rust/tests/integration/mod.rs b/integration/rust/tests/integration/mod.rs index 03badb9eb..082df0d21 100644 --- a/integration/rust/tests/integration/mod.rs +++ b/integration/rust/tests/integration/mod.rs @@ -1,4 +1,5 @@ pub mod admin; +pub mod admin_reload; pub mod admin_termination; pub mod auth; pub mod auto_id; diff --git a/pgdog/src/admin/force_reload.rs b/pgdog/src/admin/force_reload.rs new file mode 100644 index 000000000..cd7cf9047 --- /dev/null +++ b/pgdog/src/admin/force_reload.rs @@ -0,0 +1,24 @@ +//! FORCE RELOAD command. + +use super::prelude::*; +use crate::backend::databases::reload; + +pub(crate) struct ForceReload; + +#[async_trait] +impl Command for ForceReload { + fn name(&self) -> String { + "FORCE_RELOAD".into() + } + + fn parse(_sql: &str) -> Result { + Ok(ForceReload) + } + + async fn execute(&self) -> Result, Error> { + // true = force reload + reload(true)?; + + Ok(vec![]) + } +} diff --git a/pgdog/src/admin/mod.rs b/pgdog/src/admin/mod.rs index 6565801da..381647da8 100644 --- a/pgdog/src/admin/mod.rs +++ b/pgdog/src/admin/mod.rs @@ -8,6 +8,7 @@ pub(crate) mod ban; pub(crate) mod copy_data; pub(crate) mod cutover; pub(crate) mod error; +pub(crate) mod force_reload; pub(crate) mod healthcheck; pub(crate) mod maintenance_mode; pub(crate) mod named_row; @@ -55,6 +56,7 @@ pub(crate) use ban::*; pub(crate) use copy_data::*; pub(crate) use cutover::*; pub(crate) use error::Error; +pub(crate) use force_reload::*; pub(crate) use healthcheck::*; pub(crate) use maintenance_mode::*; pub(crate) use pause::*; diff --git a/pgdog/src/admin/parser.rs b/pgdog/src/admin/parser.rs index 771afa22c..545111420 100644 --- a/pgdog/src/admin/parser.rs +++ b/pgdog/src/admin/parser.rs @@ -1,8 +1,7 @@ //! Admin command parser. -use crate::admin::show_guc::get_show_variable; - use super::*; +use crate::admin::show_guc::get_show_variable; use tracing::debug; @@ -12,6 +11,7 @@ pub(crate) enum ParseResult { Reconnect(Reconnect), ShowClients(ShowClients), Reload(Reload), + ForceReload(ForceReload), ShowPools(ShowPools), ShowBans(ShowBans), ShowConfig(ShowConfig), @@ -61,6 +61,7 @@ impl ParseResult { Reconnect(reconnect) => reconnect.execute().await, ShowClients(show_clients) => show_clients.execute().await, Reload(reload) => reload.execute().await, + ForceReload(force_reload) => force_reload.execute().await, ShowPools(show_pools) => show_pools.execute().await, ShowBans(show_bans) => show_bans.execute().await, ShowConfig(show_config) => show_config.execute().await, @@ -110,6 +111,7 @@ impl ParseResult { Reconnect(reconnect) => reconnect.name(), ShowClients(show_clients) => show_clients.name(), Reload(reload) => reload.name(), + ForceReload(force_reload) => force_reload.name(), ShowPools(show_pools) => show_pools.name(), ShowBans(show_bans) => show_bans.name(), ShowConfig(show_config) => show_config.name(), @@ -206,6 +208,7 @@ impl Parser { "shutdown" => ParseResult::Shutdown(Shutdown::parse(&sql)?), "reconnect" => ParseResult::Reconnect(Reconnect::parse(&sql)?), "reload" => ParseResult::Reload(Reload::parse(&sql)?), + "force_reload" => ParseResult::ForceReload(ForceReload::parse(&sql)?), "ban" | "unban" => ParseResult::Ban(Ban::parse(&sql)?), "healthcheck" => ParseResult::Healthcheck(Healthcheck::parse(&sql)?), // These are not covered by the show handler above @@ -286,6 +289,7 @@ mod tests { assert_parses!("RESUME", ParseResult::Pause(_)); assert_parses!("RECONNECT", ParseResult::Reconnect(_)); assert_parses!("RELOAD", ParseResult::Reload(_)); + assert_parses!("FORCE_RELOAD", ParseResult::ForceReload(_)); assert_parses!("SHUTDOWN", ParseResult::Shutdown(_)); assert_parses!("BAN", ParseResult::Ban(_)); assert_parses!("UNBAN", ParseResult::Ban(_)); diff --git a/pgdog/src/admin/reload.rs b/pgdog/src/admin/reload.rs index e43c8a67a..3acee2a6a 100644 --- a/pgdog/src/admin/reload.rs +++ b/pgdog/src/admin/reload.rs @@ -16,7 +16,8 @@ impl Command for Reload { } async fn execute(&self) -> Result, Error> { - reload()?; + // false = NOT a force reload + reload(false)?; Ok(vec![]) } } diff --git a/pgdog/src/backend/databases.rs b/pgdog/src/backend/databases.rs index 08f67a164..be0729749 100644 --- a/pgdog/src/backend/databases.rs +++ b/pgdog/src/backend/databases.rs @@ -1,9 +1,5 @@ //! Databases behind pgDog. -use std::collections::HashMap; -use std::ops::Deref; -use std::sync::Arc; - use arc_swap::ArcSwap; use futures::future::try_join_all; use indexmap::IndexMap; @@ -17,6 +13,9 @@ use pgdog_config::{ EnumeratedDatabase, QueryParser, ShardedMappingConfig, ShardedMappingKey, ShardedMappingKeyRef, ShardedMappingKindDeprecated, ShardedMappingList, ShardedMappingRange, ShardedTableConfig, }; +use std::collections::HashMap; +use std::ops::Deref; +use std::sync::Arc; use tracing::{debug, error, info, warn}; use crate::auth::AuthResult; @@ -133,15 +132,32 @@ pub(crate) async fn cancel_all(database: &str) -> Result<(), Error> { Ok(()) } +/// Terminates all active connections on all `Pool`s. +pub(crate) fn terminate_active_connections() { + databases() + .all() + .values() + .for_each(Cluster::terminate_active_connections); +} + /// Re-create pools from config. -pub(crate) fn reload() -> Result<(), Error> { - info!("reloading configuration"); +pub(crate) fn reload(force: bool) -> Result<(), Error> { + if force { + info!("force reloading configuration"); + } else { + info!("reloading configuration"); + } // Load config from disk. let old_config = config(); let new_config = load(&old_config.config_path, &old_config.users_path)?; let databases = from_config(&new_config); + // Terminate after checking config for validity. + if force { + terminate_active_connections(); + } + // Replace databases. replace_databases(databases, true)?; diff --git a/pgdog/src/backend/pool/cluster.rs b/pgdog/src/backend/pool/cluster.rs index 53fd035f0..b603e4c32 100644 --- a/pgdog/src/backend/pool/cluster.rs +++ b/pgdog/src/backend/pool/cluster.rs @@ -7,6 +7,7 @@ use pgdog_config::{ users::PasswordKind, }; use std::{sync::Arc, time::Duration}; +use tokio_util::sync::CancellationToken; use crate::backend::schema::SchemaCache; use crate::backend::server::ServerRequest; @@ -87,6 +88,7 @@ pub(crate) struct Cluster { canonical_oids: Option>, read_only: bool, failover_signal: ClusterFailoverSignalWatcher, + cancellation_token: CancellationToken, } /// Bare test clusters carry the same defaults the config would apply, @@ -136,6 +138,7 @@ impl Default for Cluster { canonical_oids: Default::default(), read_only: Default::default(), failover_signal: ClusterFailoverSignalWatcher::default(), + cancellation_token: Default::default(), } } } @@ -415,9 +418,25 @@ impl Cluster { canonical_oids, read_only, failover_signal, + cancellation_token: Default::default(), } } + pub(crate) fn get_cancellation_token(&self) -> CancellationToken { + self.cancellation_token.clone() + } + + /// Terminates all active connections for the `Cluster` + /// and marks all `Pool`s as offline to refuse future connections. + pub(crate) fn terminate_active_connections(&self) { + for shard in self.shards() { + for pool in shard.pools() { + pool.set_offline(); + } + } + self.cancellation_token.cancel(); + } + /// Change config to work with logical replication streaming. pub(crate) fn logical_stream(&self) -> Self { let mut cluster = self.clone(); diff --git a/pgdog/src/backend/pool/connection/mod.rs b/pgdog/src/backend/pool/connection/mod.rs index 0c41d97c6..dbc1a01a5 100644 --- a/pgdog/src/backend/pool/connection/mod.rs +++ b/pgdog/src/backend/pool/connection/mod.rs @@ -1,8 +1,10 @@ //! Server connection requested by a frontend. +use futures::future::try_join_all; use mirror::MirrorHandler; use pgdog_config::users::PasswordKind; use tokio::select; +use tokio_util::sync::CancellationToken; use tracing::debug; use crate::{ @@ -22,7 +24,7 @@ use crate::{ }; use super::{ - super::{Error, pool::Guard}, + super::{Error, Server, pool::Guard}, Address, Cluster, Request, }; @@ -51,6 +53,9 @@ pub(crate) struct Connection { database: String, binding: Binding, cluster: Option, + /// Each client polls own child node instead of contending on the shared `Cluster` node. + /// Cancelled when an admin terminates the cluster (`FORCE_RELOAD`) + cancellation_token: CancellationToken, mirrors: Vec, pub_sub: PubSubClient, } @@ -65,6 +70,7 @@ impl Connection { Binding::NotConnected }, cluster: None, + cancellation_token: CancellationToken::new(), user: user.to_owned(), database: database.to_owned(), mirrors: vec![], @@ -396,6 +402,7 @@ impl Connection { let databases = databases(); let cluster = databases.cluster(user)?; + self.cancellation_token = cluster.get_cancellation_token().child_token(); self.cluster = Some(cluster.clone()); let source_db = cluster.name(); self.mirrors = databases @@ -466,6 +473,29 @@ impl Connection { }) } + /// Cancel the query the server(s) are running for this client + pub(crate) async fn cancel_query(&self) -> Result<(), Error> { + let servers: Vec<&Guard> = match self.binding { + Binding::Direct(ref server, ..) => vec![server], + Binding::MultiShard(ref servers, _) => servers.iter().collect(), + _ => return Ok(()), + }; + + try_join_all( + servers + .iter() + .map(|server| Server::cancel(server.addr(), server.key().clone())), + ) + .await?; + + Ok(()) + } + + /// Token cancelled when an admin terminates this connection's `Cluster`. + pub(crate) fn cancellation_token(&self) -> CancellationToken { + self.cancellation_token.clone() + } + /// Get cluster if any. #[inline] pub(crate) fn cluster(&self) -> Result<&Cluster, Error> { diff --git a/pgdog/src/backend/pool/pool_impl.rs b/pgdog/src/backend/pool/pool_impl.rs index 1c351dd74..e78236057 100644 --- a/pgdog/src/backend/pool/pool_impl.rs +++ b/pgdog/src/backend/pool/pool_impl.rs @@ -313,11 +313,9 @@ impl Pool { let mut to_guard = destination.lock(); // Propagate pause state so a paused database stays paused after reload. - if from_guard.paused { - to_guard.paused = true; - } - + to_guard.paused = from_guard.paused; from_guard.online = false; + let (idle, taken) = from_guard.move_conns_to(destination); for server in idle { to_guard.put(server, now)?; @@ -338,9 +336,8 @@ impl Pool { /// Pause pool, closing all open connections. pub(crate) fn pause(&self) { let mut guard = self.lock(); - - guard.paused = true; guard.dump_idle(); + guard.paused = true; } /// Send a cancellation request for all running queries. @@ -352,6 +349,7 @@ impl Pool { .cancel_keys() .map(|key| Server::cancel(&addr, key.clone())) .collect(); + try_join_all(futures) .await .map_err(|_| Error::FastShutdown)?; @@ -395,6 +393,13 @@ impl Pool { self.comms().ready.notify_waiters(); } + /// Sets the `Pool` offline (to refuse more connections) + /// Does not dump idle connections or shutdown. + pub(crate) fn set_offline(self) { + let mut guard = self.lock(); + guard.online = false; + } + /// Pool exclusive lock. #[inline] pub(super) fn lock(&self) -> MutexGuard<'_, RawMutex, Inner> { diff --git a/pgdog/src/frontend/client/mod.rs b/pgdog/src/frontend/client/mod.rs index cbfd65702..9d35b194c 100644 --- a/pgdog/src/frontend/client/mod.rs +++ b/pgdog/src/frontend/client/mod.rs @@ -10,6 +10,7 @@ use std::time::{Duration, Instant}; use pgdog_config::users::PasswordKind; use timeouts::Timeouts; use tokio::{select, spawn}; +use tokio_util::sync::CancellationToken; use tracing::{Level as LogLevel, debug, enabled, error, info, trace, warn}; use super::{ClientRequest, Error, PreparedStatements}; @@ -527,6 +528,8 @@ impl Client { let client_state = query_engine.client_state(); + let cancellation_token = query_engine.cancellation_token(); + select! { _ = shutdown.cancelled(), if !offline => { continue; // Wake up task. @@ -538,7 +541,7 @@ impl Client { self.server_message(&mut query_engine, message).await?; } - buffer = self.buffer(client_state) => { + buffer = self.buffer(client_state, &cancellation_token) => { let event = buffer?; // Only send requests to the backend if they are complete. @@ -637,7 +640,11 @@ impl Client { /// /// This ensures we don't check out a connection from the pool until the client /// sent a complete request. - async fn buffer(&mut self, state: State) -> Result { + async fn buffer( + &mut self, + state: State, + cancellation_token: &CancellationToken, + ) -> Result { self.client_request.clear(); // Only start timer once we receive the first message. @@ -658,23 +665,35 @@ impl Client { .timeouts .client_idle_timeout(&state, &self.client_request); - let message = - match safe_timeout(idle_timeout, self.stream_buffer.read(&mut self.stream)).await { - Err(_) => { - self.stream - .fatal(ErrorResponse::client_idle_timeout(idle_timeout, &state)) - .await?; - return Ok(BufferEvent::DisconnectAbrupt); - } + let message = select! { + message = safe_timeout(idle_timeout, self.stream_buffer.read(&mut self.stream)) => { + message + } + // If any of the `CancellationTokens `trigger, exit early. Currently used for admin `FORCE_RELOAD`. + // If this returns an Error, it'll be propagated up to `Client`'s [`Box::pin(self.run())`] + // which will disconnect the `Client` (and `QueryEngine` transactions) + _ = cancellation_token.cancelled() => { + return Err(Error::AdminTermination) + } + }; - Ok(Ok(message)) => message.stream(self.streaming).frontend(), - Ok(Err(err)) => { - if let Some(response) = err.as_fatal_error_response() { - self.stream.fatal(response).await?; - } - return Ok(BufferEvent::DisconnectAbrupt); + let message = match message { + Err(_) => { + self.stream + .fatal(ErrorResponse::client_idle_timeout(idle_timeout, &state)) + .await?; + return Ok(BufferEvent::DisconnectAbrupt); + } + + Ok(Ok(message)) => message.stream(self.streaming).frontend(), + + Ok(Err(err)) => { + if let Some(response) = err.as_fatal_error_response() { + self.stream.fatal(response).await?; } - }; + return Ok(BufferEvent::DisconnectAbrupt); + } + }; if timer.is_none() { timer = Some(Instant::now()); diff --git a/pgdog/src/frontend/client/query_engine/mod.rs b/pgdog/src/frontend/client/query_engine/mod.rs index 4ae49f5d0..9196bcf8b 100644 --- a/pgdog/src/frontend/client/query_engine/mod.rs +++ b/pgdog/src/frontend/client/query_engine/mod.rs @@ -10,6 +10,7 @@ use crate::{ net::{ErrorResponse, Message, Parameters}, state::State, }; +use tokio_util::sync::CancellationToken; use tracing::debug; pub(crate) mod advisory_lock; @@ -109,6 +110,11 @@ impl QueryEngine { Self::new(&client.params, &client.comms, client.admin) } + /// Token cancelled when an admin terminates this client's cluster (FORCE_RELOAD). + pub(crate) fn cancellation_token(&self) -> CancellationToken { + self.backend.cancellation_token() + } + /// Wait for an async message from the backend. pub(crate) async fn read_backend(&mut self) -> Result { Ok(self.backend.read().await?) diff --git a/pgdog/src/frontend/client/query_engine/query.rs b/pgdog/src/frontend/client/query_engine/query.rs index f7f3987af..aa88e8f99 100644 --- a/pgdog/src/frontend/client/query_engine/query.rs +++ b/pgdog/src/frontend/client/query_engine/query.rs @@ -65,12 +65,30 @@ impl QueryEngine { } } + let cancellation_token = self.backend.cancellation_token(); + let query_timeout = context.timeouts.query_timeout(&State::Active); - let result = safe_timeout( - query_timeout, - self.client_server_exchange(context, query_planner), - ) - .await; + + let result = tokio::select! { + result = safe_timeout( + query_timeout, + self.client_server_exchange(context, query_planner), + ) => { + result + } + // If the cluster's cancellation token triggers, exit early. Currently used for admin FORCE_RELOAD. + // If this returns an Error, it'll be propagated up to Client's Box::pin(self.run()) + // which will disconnect the client (and QueryEngine transactions) + _ = cancellation_token.cancelled() => { + // Postgres is still running the query. Send a cancellation request before we stop on our end. + if let Err(err) = self.backend.cancel_query().await { + // Tell the administrator that we failed to cancel the query. + error!("failed to cancel query on admin termination: {err}"); + } + self.backend.force_close(); + return Err(Error::AdminTermination); + } + }; match result { Ok(response) => response?, diff --git a/pgdog/src/frontend/client/test/mod.rs b/pgdog/src/frontend/client/test/mod.rs index 5b9906827..febb77a5e 100644 --- a/pgdog/src/frontend/client/test/mod.rs +++ b/pgdog/src/frontend/client/test/mod.rs @@ -1,4 +1,5 @@ use std::time::{Duration, Instant}; +use tokio_util::sync::CancellationToken; use pgdog_config::{PoolerMode, QuerySizeLimitAction}; use tokio::{ @@ -253,7 +254,10 @@ async fn test_abrupt_disconnect() { drop(conn); - let event = client.buffer(State::Idle).await.unwrap(); + let event = client + .buffer(State::Idle, &CancellationToken::new()) + .await + .unwrap(); assert_eq!(event, BufferEvent::DisconnectAbrupt); assert!(client.client_request.messages.is_empty()); @@ -272,7 +276,10 @@ async fn test_client_idle_timeout() { set(config).unwrap(); let start = Instant::now(); - let res = client.buffer(State::Idle).await.unwrap(); + let res = client + .buffer(State::Idle, &CancellationToken::new()) + .await + .unwrap(); assert_eq!(res, BufferEvent::DisconnectAbrupt); let err = read_one!(conn); @@ -283,7 +290,7 @@ async fn test_client_idle_timeout() { assert!( timeout( Duration::from_millis(50), - client.buffer(State::IdleInTransaction) + client.buffer(State::IdleInTransaction, &CancellationToken::new()) ) .await .is_err() @@ -635,7 +642,10 @@ async fn test_query_timeout() { let buf = buffer!({ Query::new("SELECT pg_sleep(0.2)") }); conn.write_all(&buf).await.unwrap(); - client.buffer(State::Idle).await.unwrap(); + client + .buffer(State::Idle, &CancellationToken::new()) + .await + .unwrap(); let result = client.client_messages(&mut engine).await; assert!(result.is_err()); diff --git a/pgdog/src/frontend/client/test/test_client.rs b/pgdog/src/frontend/client/test/test_client.rs index 11fb6214e..d4a6cbdbe 100644 --- a/pgdog/src/frontend/client/test/test_client.rs +++ b/pgdog/src/frontend/client/test/test_client.rs @@ -1,4 +1,5 @@ use std::{fmt::Debug, ops::Deref}; +use tokio_util::sync::CancellationToken; use bytes::{BufMut, Bytes, BytesMut}; use pgdog_config::RewriteMode; @@ -237,7 +238,9 @@ impl TestClient { /// Process a request. pub(crate) async fn try_process(&mut self) -> Result<(), Box> { - self.client.buffer(self.engine.stats().state).await?; + self.client + .buffer(self.engine.stats().state, &CancellationToken::new()) + .await?; self.client.client_messages(&mut self.engine).await?; Ok(()) diff --git a/pgdog/src/frontend/error.rs b/pgdog/src/frontend/error.rs index 92e6950a4..6a8ced79c 100644 --- a/pgdog/src/frontend/error.rs +++ b/pgdog/src/frontend/error.rs @@ -57,6 +57,9 @@ pub(crate) enum Error { #[error("multi-tuple insert requires multi-shard binding")] MultiShardRequired, + #[error("terminating connection due to administrator command")] + AdminTermination, + // FIXME: layer errors better so we don't have // to reach so deep into a module. #[error("{0}")] diff --git a/pgdog/src/frontend/listener.rs b/pgdog/src/frontend/listener.rs index c2d6fb8f8..967ff128d 100644 --- a/pgdog/src/frontend/listener.rs +++ b/pgdog/src/frontend/listener.rs @@ -118,7 +118,7 @@ impl Listener { } _ = sighup.listen() => { - if let Err(err) = reload() { + if let Err(err) = reload(false) { error!("configuration reload error: {}", err); } } diff --git a/pgdog/src/net/messages/error_response.rs b/pgdog/src/net/messages/error_response.rs index c90a57757..526d541a4 100644 --- a/pgdog/src/net/messages/error_response.rs +++ b/pgdog/src/net/messages/error_response.rs @@ -207,6 +207,19 @@ impl ErrorResponse { } } + /// Terminating due to admin command (e.g. FORCE_RELOAD) + pub(crate) fn admin_termination() -> ErrorResponse { + ErrorResponse { + severity: "FATAL".into(), + code: "57P01".into(), + message: "terminating connection due to administrator command".into(), + detail: None, + context: None, + file: None, + routine: None, + } + } + pub(crate) fn syntax>(err: T) -> ErrorResponse { Self { severity: "ERROR".into(), @@ -260,6 +273,9 @@ impl ErrorResponse { use crate::backend::Error as BackendError; if let FrontendError::Backend(BackendError::ExecutionError(err)) = err { *(err.clone()) + } else if let FrontendError::AdminTermination = err { + // Allows us to set a custom code (to identically represent the same Postgres error) + ErrorResponse::admin_termination() } else { Self { severity: "FATAL".into(),