From 70f8e27922796de3ef5f8b02eaa9d788fa9d65ff Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ra=C3=BAl=20Piaggio?= Date: Fri, 3 Jul 2026 02:28:26 -0300 Subject: [PATCH 1/3] Fail notification streams when the connection is lost --- .../scala/net/BufferedMessageSocket.scala | 20 +++++++++++++------ 1 file changed, 14 insertions(+), 6 deletions(-) diff --git a/modules/core/shared/src/main/scala/net/BufferedMessageSocket.scala b/modules/core/shared/src/main/scala/net/BufferedMessageSocket.scala index e05da2fb6..b4a7b95c4 100644 --- a/modules/core/shared/src/main/scala/net/BufferedMessageSocket.scala +++ b/modules/core/shared/src/main/scala/net/BufferedMessageSocket.scala @@ -94,10 +94,11 @@ object BufferedMessageSocket { */ private def next[F[_]: MonadThrow]( ms: MessageSocket[F], + term: Ref[F, Option[Throwable]], xaSig: Ref[F, TransactionStatus], paSig: Ref[F, Map[String, String]], bkDef: Deferred[F, BackendKeyData], - noTop: Topic[F, Notification[String]], + noTop: Topic[F, Either[Throwable, Notification[String]]], queue: Queue[F, BackendMessage] ): F[Unit] = { def step: F[Unit] = ms.receive.flatMap { @@ -108,15 +109,18 @@ object BufferedMessageSocket { case m @ ReadyForQuery(s) => xaSig.set(s) >> queue.offer(m) // observe and then emit // These are handled here and are never seen by the higher-level API. case ParameterStatus(k, v) => paSig.update(_ + (k -> v)) - case NotificationResponse(n) => noTop.publish1(n).void // TODO -- what if it's closed? + case NotificationResponse(n) => noTop.publish1(Right(n)).void // topic only closes after a terminal error, at which point dropping is correct case NoticeResponse(_) => Monad[F].unit // TODO -- we're throwing these away! case m @ BackendKeyData(_, _) => bkDef.complete(m).void // Everything else is passed through. case m => queue.offer(m) } >> step + // Publish the failure to synchronous exchanges (via the queue) and to notification + // subscribers (via the topic, which is then closed so `listen` streams terminate + // instead of hanging silently on a dead connection). step.attempt.flatMap { - case Left(e) => queue.offer(NetworkError(e)) // publish the failure + case Left(e) => term.set(Some(e)) *> queue.offer(NetworkError(e)) *> noTop.publish1(Left(e)) *> noTop.close.void case Right(_) => Monad[F].unit } } @@ -134,8 +138,8 @@ object BufferedMessageSocket { xaSig <- SignallingRef[F, TransactionStatus](TransactionStatus.Idle) // initial state (ok) paSig <- SignallingRef[F, Map[String, String]](Map.empty) bkSig <- Deferred[F, BackendKeyData] - noTop <- Topic[F, Notification[String]] - fib <- next(ms, xaSig, paSig, bkSig, noTop, queue).start + noTop <- Topic[F, Either[Throwable, Notification[String]]] + fib <- next(ms, term, xaSig, paSig, bkSig, noTop, queue).start } yield new AbstractMessageSocket[F] with BufferedMessageSocket[F] { @@ -161,7 +165,11 @@ object BufferedMessageSocket { override def backendKeyData: Deferred[F, BackendKeyData] = bkSig override def notifications(maxQueued: Int): Resource[F, Stream[F, Notification[String]]] = - noTop.subscribeAwait(maxQueued) + noTop.subscribeAwait(maxQueued).map { s => + // The topic closes after the terminal error is published; the trailing check covers + // subscribers that arrive after the failure and would otherwise see an empty stream. + s.rethrow ++ Stream.exec(term.get.flatMap(_.traverse_(Concurrent[F].raiseError[Unit](_)))) + } override protected def terminate: F[Unit] = fib.cancel *> // stop processing incoming messages From 727fd226df319165d91828db17ca98e3236eccff Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ra=C3=BAl=20Piaggio?= Date: Fri, 3 Jul 2026 13:42:52 -0300 Subject: [PATCH 2/3] add test for disconnection notification --- .../shared/src/test/scala/DisconnectTest.scala | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/modules/tests/shared/src/test/scala/DisconnectTest.scala b/modules/tests/shared/src/test/scala/DisconnectTest.scala index 25ec3f060..d48827096 100644 --- a/modules/tests/shared/src/test/scala/DisconnectTest.scala +++ b/modules/tests/shared/src/test/scala/DisconnectTest.scala @@ -4,6 +4,8 @@ package tests +import cats.effect._ +import scala.concurrent.duration._ import skunk.implicits._ import skunk.codec.all._ import skunk.exception.EofException @@ -19,4 +21,19 @@ class DisconnectTest extends SkunkTest { } } + pooledTest("listen fails when the connection is lost", max = 1) { p => + p.use { s => // this session will be invalidated, so its release fails too + for { + fib <- s.channel(id"disconnect_test").listen(42).compile.drain.start + _ <- IO.sleep(1.second) // give the fiber time to issue LISTEN (see the race note in ChannelTest) + _ <- s.execute(sql"select pg_terminate_backend(pg_backend_pid())".query(bool)).assertFailsWith[EofException] + oc <- fib.join.timeout(10.seconds) // hangs forever if the failure is not propagated to the stream + _ <- oc match { + case Outcome.Errored(_) => IO.unit + case o => fail[Unit](s"expected listen stream to fail, got $o") + } + } yield () + }.assertFailsWith[EofException] *> IO.pure("ok") + } + } From 620b9eca08a4cd5a3b58a90eccc9a74cad078770 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ra=C3=BAl=20Piaggio?= Date: Sun, 16 Aug 2026 13:10:27 +0200 Subject: [PATCH 3/3] fix race condition --- .../main/scala/net/BufferedMessageSocket.scala | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/modules/core/shared/src/main/scala/net/BufferedMessageSocket.scala b/modules/core/shared/src/main/scala/net/BufferedMessageSocket.scala index b4a7b95c4..381b90075 100644 --- a/modules/core/shared/src/main/scala/net/BufferedMessageSocket.scala +++ b/modules/core/shared/src/main/scala/net/BufferedMessageSocket.scala @@ -94,7 +94,7 @@ object BufferedMessageSocket { */ private def next[F[_]: MonadThrow]( ms: MessageSocket[F], - term: Ref[F, Option[Throwable]], + noErr: Ref[F, Option[Throwable]], xaSig: Ref[F, TransactionStatus], paSig: Ref[F, Map[String, String]], bkDef: Deferred[F, BackendKeyData], @@ -116,11 +116,12 @@ object BufferedMessageSocket { case m => queue.offer(m) } >> step - // Publish the failure to synchronous exchanges (via the queue) and to notification - // subscribers (via the topic, which is then closed so `listen` streams terminate - // instead of hanging silently on a dead connection). + // Publish the failure to synchronous exchanges (via the queue, behind any messages already + // buffered, so in-flight exchanges still see them) and to notification subscribers (via the + // topic, which is then closed so `listen` streams terminate instead of hanging silently on a + // dead connection). step.attempt.flatMap { - case Left(e) => term.set(Some(e)) *> queue.offer(NetworkError(e)) *> noTop.publish1(Left(e)) *> noTop.close.void + case Left(e) => noErr.set(Some(e)) *> queue.offer(NetworkError(e)) *> noTop.publish1(Left(e)) *> noTop.close.void case Right(_) => Monad[F].unit } } @@ -134,12 +135,13 @@ object BufferedMessageSocket { ): F[BufferedMessageSocket[F]] = for { term <- Ref[F].of[Option[Throwable]](None) // terminal error + noErr <- Ref[F].of[Option[Throwable]](None) // terminal error for notification subscribers queue <- Queue.bounded[F, BackendMessage](queueSize) xaSig <- SignallingRef[F, TransactionStatus](TransactionStatus.Idle) // initial state (ok) paSig <- SignallingRef[F, Map[String, String]](Map.empty) bkSig <- Deferred[F, BackendKeyData] noTop <- Topic[F, Either[Throwable, Notification[String]]] - fib <- next(ms, term, xaSig, paSig, bkSig, noTop, queue).start + fib <- next(ms, noErr, xaSig, paSig, bkSig, noTop, queue).start } yield new AbstractMessageSocket[F] with BufferedMessageSocket[F] { @@ -168,7 +170,7 @@ object BufferedMessageSocket { noTop.subscribeAwait(maxQueued).map { s => // The topic closes after the terminal error is published; the trailing check covers // subscribers that arrive after the failure and would otherwise see an empty stream. - s.rethrow ++ Stream.exec(term.get.flatMap(_.traverse_(Concurrent[F].raiseError[Unit](_)))) + s.rethrow ++ Stream.exec(noErr.get.flatMap(_.traverse_(Concurrent[F].raiseError[Unit](_)))) } override protected def terminate: F[Unit] =