fix: forward LISTEN/NOTIFY/UNLISTEN to backend in session mode

integration/complex/run.sh

@@ -6,4 +6,5 @@ pushd ${SCRIPT_DIR}
 bash shutdown.sh
 bash passthrough_auth/run.sh
 bash cancel_query/run.sh
+bash session_listen/run.sh
 popd

integration/complex/session_listen/pgdog.toml

@@ -0,0 +1,14 @@
+[general]
+pub_sub_channel_size = 0
+
+[rewrite]
+enabled = false
+shard_key = "ignore"
+split_inserts = "error"
+
+[[databases]]
+name = "pgdog"
+host = "127.0.0.1"
+
+[admin]
+password = "pgdog"

integration/complex/session_listen/run.sh

@@ -0,0 +1,30 @@
+#!/bin/bash
+#
+# Regression test: LISTEN must not be rejected with "pub/sub disabled"
+# when pub_sub_channel_size = 0 and the user is in session mode.
+#
+set -e
+SCRIPT_DIR=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd )
+
+export PGPASSWORD=pgdog
+
+PGDOG_BIN_PATH="${PGDOG_BIN:-${SCRIPT_DIR}/../../../target/release/pgdog}"
+
+killall -TERM pgdog 2> /dev/null || true
+
+"${PGDOG_BIN_PATH}" \
+    --config ${SCRIPT_DIR}/pgdog.toml \
+    --users ${SCRIPT_DIR}/users.toml &
+PGDOG_PID=$!
+
+until pg_isready -h 127.0.0.1 -p 6432 -U pgdog_session -d pgdog; do
+    sleep 1
+done
+
+psql -h 127.0.0.1 -p 6432 -U pgdog_session -d pgdog -c "LISTEN test_channel"
+psql -h 127.0.0.1 -p 6432 -U pgdog_session -d pgdog -c "UNLISTEN *"
+
+echo "PASS: session mode LISTEN/UNLISTEN with pub_sub disabled"
+
+killall -TERM pgdog
+wait "${PGDOG_PID}" 2> /dev/null || true

integration/complex/session_listen/users.toml

@@ -0,0 +1,6 @@
+[[users]]
+name = "pgdog_session"
+database = "pgdog"
+password = "pgdog"
+server_user = "pgdog"
+pooler_mode = "session"

integration/rust/tests/integration/notify.rs

@@ -378,3 +378,13 @@ async fn test_notify_not_delivered_after_constraint_violation() {
     let _ = conn.execute("DROP TABLE test_notify_constraint").await;
     listener_task.abort();
 }
+
+#[tokio::test]
+async fn test_listen_session_mode() {
+    let mut conn = PgConnection::connect("postgres://pgdog_session:pgdog@127.0.0.1:6432/pgdog")
+        .await
+        .unwrap();
+
+    conn.execute("LISTEN test_session_channel").await.unwrap();
+    conn.execute("UNLISTEN test_session_channel").await.unwrap();
+}

pgdog/src/frontend/client/query_engine/mod.rs

@@ -230,6 +230,11 @@ impl QueryEngine {
                 context.params.rollback();
             }
             Command::Query(_) => self.execute(context).await?,
+            Command::Listen { .. } | Command::Notify { .. } | Command::Unlisten(_)
+                if self.backend.session_mode() =>
+            {
+                self.execute(context).await?
+            }
             Command::Listen { channel, shard } => {
                 self.listen(context, &channel.clone(), shard.clone())
                     .await?