fix(client,pgpool): read idle socket messages and isolate connection failures

Author: myfreess

client/INTERNAL.md

@@ -53,9 +53,12 @@ temporary prepared-statement cleanup.
 
 ## Driver And Queues
 
-The driver processes one `Request` at a time:
+The driver keeps one reader task on the socket for the entire established
+connection, including idle periods. It routes notices, notifications, and
+parameter updates immediately. Ordinary backend responses pass through a
+one-message queue to the request loop, which processes one `Request` at a time:
 
-1. receive and retain the next accepted operation request before any socket I/O
+1. receive and retain the next accepted operation request
 2. write its frontend bytes
 3. route every ordinary response to that request's bounded response queue
 4. continue until `ReadyForQuery` or the request's protocol-specific terminal
@@ -65,15 +68,15 @@ The driver processes one `Request` at a time:
 There is no pending-request list, opportunistic pipeline decision, or
 oldest-request response routing.
 
-Each request response queue remains bounded (currently eight messages).
-Backpressure therefore stops socket reads for the active request rather than
-growing memory without limit. Because execution is single-flight, a stalled
-consumer cannot be bypassed by a later request.
+Each request response queue remains bounded (currently eight messages). The
+one-message reader handoff keeps socket reads bounded when an active response
+consumer stalls. Because execution is single-flight, a later request cannot
+bypass that consumer.
 
 COPY IN is the protocol exception that requires bidirectional coordination.
 Its producer queue is bounded to eight actions. After PostgreSQL enters COPY
 mode, the driver starts exactly one request-local writer for data, finish, or
-abort actions while the driver itself remains the sole backend reader. An early
+abort actions while its reader task remains the sole backend reader. An early
 `ErrorResponse` is forwarded to the sink, the input queue is cleared and closed,
 and the writer stops after its current complete frame. The driver joins that
 writer before forwarding the final `ReadyForQuery` or starting another request.
@@ -192,7 +195,8 @@ Out-of-band backend messages never enter an operation response queue:
 
 They update shared state and are published through the Client async-message
 queue. `Client::next_message()` has one-consumer semantics and returns `None`
-when the driver terminates.
+when the driver terminates. The reader continues delivering these messages
+without a query in flight.
 
 Before starting a physical connection, `pgpool` installs the internal async
 message handler. That handler forwards messages to the pool's buffer instead

client/INTERNAL_CN.md

@@ -44,23 +44,25 @@
 
 ## 驱动器与队列
 
-驱动器每次处理一个 `Request`：
+驱动器在连接建立后的整个生命周期维持一个套接字读取任务，包括空闲时段。
+提示、通知和参数更新会立即发布；普通响应经容量为一的队列交给请求循环。
+请求循环每次处理一个 `Request`：
 
-1. 在进行任何套接字 I/O 前，先接收并保留下一个已接受操作的请求
+1. 接收并保留下一个已接受操作的请求
 2. 写入该请求的前端协议字节
 3. 将每条普通响应路由到该请求的有界响应队列
 4. 持续处理，直到收到 `ReadyForQuery` 或到达该请求协议规定的终止状态
 5. 转而处理下一个请求
 
 驱动器不维护待处理请求列表，不会择机启用流水线，也不会按最早请求来路由响应。
 
-每个请求的响应队列始终有容量上限（目前为八条消息）。因此，背压会暂停活动请求的
-套接字读取，而不会让内存无限增长。由于同一时刻只执行一个操作，后续请求无法绕过
-停滞的消费者。
+每个请求的响应队列始终有容量上限（目前为八条消息）。读取任务到请求循环的队列
+最多缓存一条消息，活动响应的消费者停滞时，背压仍会限制套接字读取和内存占用。
+由于同一时刻只执行一个操作，后续请求无法绕过停滞的消费者。
 
 COPY IN 是协议中的例外，需要双向协调。其生产者队列最多容纳八个动作。PostgreSQL
 进入 COPY 模式后，驱动器会为该请求创建且仅创建一个写入任务，用来处理数据、完成或
-中止动作，而驱动器自身仍是唯一的后端消息读取者。提前到达的 `ErrorResponse` 会被
+中止动作，而读取任务仍是唯一的后端消息读取者。提前到达的 `ErrorResponse` 会被
 转发给写入端（sink），输入队列随即被清空并关闭，写入任务则在完成当前整个帧的写入后
 停止。驱动器会等待该写入任务结束，然后才转发最终的 `ReadyForQuery` 或开始另一个请求。
 整个过程仍然属于同一个逻辑操作。
@@ -142,7 +144,8 @@ async 0.22 在离开受保护代码块时，不会自动传播待处理的取消
 - `ParameterStatus`
 
 它们会更新共享状态，并通过 Client 的异步消息队列发布。`Client::next_message()`
-采用单消费者语义，在驱动器终止时返回 `None`。
+采用单消费者语义，在驱动器终止时返回 `None`。即使没有查询在执行，读取任务也会
+持续交付这些消息。
 
 `pgpool` 在启动物理连接前安装内部异步消息处理器，将消息转发到连接池的缓冲队列，
 而不是 Client 队列；连接池通过 `Pool::next_message()` 消费这些消息。

client/README.mbt.md

@@ -353,7 +353,8 @@ async fn _read_async_message(client : @client.Client) -> @client.AsyncMessage? {
 }
 ```
 
-Use one dedicated consumer for `next_message()`. It returns `None` after the
+Use one dedicated consumer for `next_message()`. Notifications arrive while the
+connection is idle, without sending another query. It returns `None` after the
 driver closes, after buffered messages are read. By default the client retains
 at most 256 messages. `Client::create_with_async_message_capacity(config,
 capacity)` accepts a positive capacity. When full, the driver discards the

client/connection_idle_wbtest.mbt

@@ -0,0 +1,82 @@
+///|
+fn idle_notification_packet(
+  channel : String,
+  payload : String,
+) -> Bytes raise @proto.ProtocolError {
+  let body = Buffer()
+  body.write_int_be(123)
+  body.write_bytes(@proto.utf8_encode(channel))
+  body.write_byte(0)
+  body.write_bytes(@proto.utf8_encode(payload))
+  body.write_byte(0)
+  connect_backend_packet(@backend.NOTIFICATION_RESPONSE_TAG, body)
+}
+
+///|
+async test "idle connection delivers notifications without another request" {
+  @async.with_task_group(group => {
+    let (client, connection, backend) = connection_test_pair(group)
+    defer backend.close()
+    let driver = group.spawn(no_wait=true, allow_failure=true, () => {
+      connection.run(group)
+    })
+    client.shared.driver_task.val = Some(driver)
+    defer client.abort()
+    @async.with_timeout(2000, () => {
+      backend.write(idle_notification_packet("moon_idle", "first"))
+      assert_true(
+        client.next_message()
+        is Some(Notification({ channel: "moon_idle", payload: "first", .. })),
+      )
+      let ping = group.spawn(() => client.check_connection())
+      assert_eq(backend.read_exactly(5), sync_bytes())
+      backend.write(ready_for_query_packet(b'I'))
+      ping.wait()
+    })
+  })
+}
+
+///|
+async test "partial idle notification stays aligned when a request arrives" {
+  @async.with_task_group(group => {
+    let (client, connection, backend) = connection_test_pair(group)
+    defer backend.close()
+    let driver = group.spawn(no_wait=true, allow_failure=true, () => {
+      connection.run(group)
+    })
+    client.shared.driver_task.val = Some(driver)
+    defer client.abort()
+    @async.with_timeout(2000, () => {
+      let notification = idle_notification_packet("moon_partial", "second")
+      backend.write(notification[:3].to_owned())
+      let ping = group.spawn(() => client.check_connection())
+      assert_eq(backend.read_exactly(5), sync_bytes())
+      backend.write(notification[3:].to_owned())
+      backend.write(ready_for_query_packet(b'I'))
+      assert_true(
+        client.next_message()
+        is Some(
+          Notification({ channel: "moon_partial", payload: "second", .. })
+        ),
+      )
+      ping.wait()
+    })
+  })
+}
+
+///|
+async test "idle socket EOF closes the driver without another request" {
+  @async.with_task_group(group => {
+    let (client, connection, backend) = connection_test_pair(group)
+    let driver = group.spawn(no_wait=true, allow_failure=true, () => {
+      connection.run(group)
+    })
+    @async.with_timeout(2000, () => {
+      backend.close()
+      driver.wait()
+      assert_true(client.is_closed())
+      assert_true(client.shared.runtime_error.val is Some(_))
+      assert_true(client.next_message() is None)
+    })
+  })
+}

client/connection_loop.mbt

@@ -12,6 +12,10 @@
 async fn Driver::run(self : Driver, group : @async.TaskGroup[Unit]) -> Unit {
   let mut active : Request? = None
   let mut cleaned_up = false
+  // The reader must survive the transition between idle time and a request.
+  // Cancelling an in-progress frame read there would lose protocol alignment.
+  let incoming : @async.Queue[@backend.Message] = Queue(kind=Blocking(1))
+  let reading_request = @ref.new(false)
   fn cleanup(error : Error?) {
     if cleaned_up {
       return
@@ -43,6 +47,15 @@ async fn Driver::run(self : Driver, group : @async.TaskGroup[Unit]) -> Unit {
       self.shared.control.close()
     }
   }
+  let reader = group.spawn(no_wait=true, allow_failure=true, () => {
+    self.receive_messages(incoming, reading_request)
+  })
+  defer @async.protect_from_cancel(() => {
+    reader.cancel()
+    reader.wait() catch {
+      _ => ()
+    }
+  })
   try {
     while true {
       let request = self.shared.requests.get()
@@ -55,8 +68,9 @@ async fn Driver::run(self : Driver, group : @async.TaskGroup[Unit]) -> Unit {
           return
         }
         Messages | CopyIn => {
+          reading_request.val = true
           self.stream.write(request.bytes)
-          self.drain_request(request, group)
+          self.drain_request(request, incoming, group)
           self.close_request(request, None)
           active = None
         }
@@ -67,6 +81,48 @@ async fn Driver::run(self : Driver, group : @async.TaskGroup[Unit]) -> Unit {
   }
 }
 
+///|
+/// Keep one owner of socket reads even when no request is in flight.
+async fn Driver::receive_messages(
+  self : Driver,
+  incoming : @async.Queue[@backend.Message],
+  reading_request : @ref.Ref[Bool],
+) -> Unit {
+  let mut failure : Error? = None
+  try {
+    while true {
+      let message = self.read_message()
+      if self.handle_async_message(message) {
+        continue
+      }
+      if !reading_request.val {
+        match message {
+          ErrorResponse(body) =>
+            raise ClientError::Database(parse_database_error(body.fields()))
+          _ =>
+            raise ClientError::UnexpectedMessage(
+              "backend response without an active request",
+            )
+        }
+      }
+      if message is ReadyForQuery(_) {
+        reading_request.val = false
+      }
+      incoming.put(message)
+    }
+  } catch {
+    error => failure = Some(error)
+  }
+  match failure {
+    Some(error) => {
+      // Preserve queued requests for Driver::cleanup to close their own queues.
+      incoming.close(error~, clear=true)
+      self.shared.requests.close(error~)
+    }
+    None => ()
+  }
+}
+
 ///|
 /// Receive the next asynchronous message buffered by the private driver.
 ///
@@ -83,17 +139,15 @@ pub async fn Client::next_message(self : Client) -> AsyncMessage? {
 async fn Driver::drain_request(
   self : Driver,
   request : Request,
+  incoming : @async.Queue[@backend.Message],
   group : @async.TaskGroup[Unit],
 ) -> Unit {
   if request.kind is CopyIn {
-    self.drain_copy_in_request(request, group)
+    self.drain_copy_in_request(request, incoming, group)
     return
   }
   while true {
-    let message = self.read_message()
-    if self.handle_async_message(message) {
-      continue
-    }
+    let message = incoming.get()
     match request.kind {
       Messages => self.forward_message_to_request(request, message)
       CopyIn => abort("COPY IN requests use drain_copy_in_request")
@@ -131,17 +185,15 @@ async fn Driver::forward_message_to_request(
 async fn Driver::drain_copy_in_request(
   self : Driver,
   request : Request,
+  incoming : @async.Queue[@backend.Message],
   group : @async.TaskGroup[Unit],
 ) -> Unit {
   let copy_input = request.copy_input.unwrap()
   let copy_error = request.copy_error.unwrap()
   let writer_stopped = @ref.new(false)
   let mut writer : @async.Task[Unit]? = None
   while true {
-    let message = self.read_message()
-    if self.handle_async_message(message) {
-      continue
-    }
+    let message = incoming.get()
     match message {
       CopyInResponse(_) => {
         guard writer is None else {

pgpool/INTERNAL.md

@@ -36,5 +36,6 @@ only then may it allocate a request ID and mark a new request active.
 Streaming callbacks finish or abandon their raw handles before returning the
 connection. Detached drains save expected connection closure for `finish()`;
 database errors remain in stream state, and unexpected protocol errors fail the
-owning client executor and then the pool executor. Ordinary task cancellation
-does not imply a PostgreSQL cancel packet, automatic retry, or physical abort.
+owning client executor. The pool retires that physical connection on return or
+its next checkout. Ordinary task cancellation does not imply a PostgreSQL
+cancel packet, automatic retry, or physical abort.

pgpool/INTERNAL_CN.md

@@ -28,5 +28,6 @@
 
 流式回调在归还连接前，会先结束或放弃其底层句柄。分离的排空任务会保存预期的连接
 关闭结果，供 `finish()` 读取；数据库错误保留在流状态中，
-意外的协议错误会使所属连接和连接池执行器失败。普通的任务取消并不意味着发送 PostgreSQL 取消报文、
+意外的协议错误会使所属客户端执行器失败；连接池在归还该连接或下一次借用时将其退役。
+普通的任务取消并不意味着发送 PostgreSQL 取消报文、
 自动重试或中止物理连接。

pgpool/README.mbt.md

@@ -289,6 +289,7 @@ fills, the oldest message is discarded and
 `Pool::dropped_async_messages()` increases. `PoolConfig::PoolConfig` accepts a
 positive `async_message_capacity`; the default is 256. After `Pool::close()`,
 buffered messages can still be read, then `next_message()` returns `None`.
+An idle listening connection delivers notifications without another checkout.
 
 Create and recycle deadlines are hard for client I/O: when either expires, the
 pool first calls `Client::abort()` on the candidate physical connection and
@@ -312,8 +313,10 @@ and await `pool.ready()` before use. A repeated `run()` raises
 `PoolError::ExecutorAlreadyStarted`. `Pool::close()` rejects new operations and
 closes idle connections. Active callbacks are allowed to finish; their physical
 connections close instead of returning to the idle set. Await the executor task
-to know that all connections have stopped. Cancellation or an unexpected
-background connection failure closes the whole pool.
+to know that all connections have stopped. Pool executor cancellation or an
+unexpected pool maintenance failure closes the whole pool. An individual
+physical connection failure closes that client; the pool discards it on return
+or on its next checkout and can create a replacement.
 
 `Pool::resize()` never interrupts checked-out sessions. When shrinking cannot
 remove enough currently available capacity tokens, the pool records resize