Reconnection bug

README.md

@@ -210,6 +210,36 @@ object Example extends IOApp {
 
 Both backends implement the same `AmqpClient[F]` trait, so switching between them should only require changing the client creation and imports.
 
+# Migrating to 4.0.3 and above — connection recovery fix (fs2-rabbit backend)
+
+Versions **4.0.0-M1 through 4.0.2** contain a connection-recovery bug that affects **`Fs2RabbitAmqpClient` only**. The `JavaBackendAmqpClient` is not affected.
+
+## What went wrong
+
+The fs2-rabbit backend ran each consumer as an fs2 `Stream` inside a `.background` fiber, but silently ignored the fiber outcome. When a network blip caused the underlying AMQP channel to close, the stream terminated with an error that was swallowed — no restart was ever attempted. After the Java AMQP client automatically recovered the connection, messages were delivered to the channel's internal buffer but nobody was draining it, so the consumer appeared permanently stuck.
+
+The v3 Java backend (and `JavaBackendAmqpClient` in v4) use a `DefaultConsumer` callback on an `AutorecoveringChannel`, which the AMQP Java client automatically re-registers after recovery. The fs2-rabbit backend had no equivalent mechanism.
+
+A secondary issue in the same code path: handler exceptions propagated through the stream and also killed it, rather than being resolved via `exceptionalAction`.
+
+## Fix in 4.0.3
+
+- The consumer loop now wraps each run in a retry: on any failure it logs a warning, waits for `networkRecoveryInterval` (default 3 seconds, matching the Java AMQP client's recovery interval), then creates a **fresh channel and consumer** before resuming. A fresh channel is required because the auto-recovered channel's dead internal fs2-rabbit queue would otherwise accumulate un-acked messages indefinitely.
+- Handler exceptions are now caught and resolved via `exceptionalAction` rather than terminating the stream.
+
+## Action required
+
+If you are using **`Fs2RabbitAmqpClient`** on any 4.x release prior to 4.0.3, upgrade to **4.0.3**:
+
+```scala
+// For fs2-rabbit Backend
+libraryDependencies += "com.itv" %% "bucky-backend-fs2-rabbit" % "4.0.3"
+```
+
+No code changes are required — the fix is entirely internal to `registerConsumer`.
+
+If you are using **`JavaBackendAmqpClient`** you are not affected and no action is needed.
+
 # Releasing a new version
 
 1. Merge your change into master

backendFs2Rabbit/src/main/scala/com/itv/bucky/backend/fs2rabbit/Fs2RabbitAmqpClient.scala

@@ -25,6 +25,7 @@ import com.itv.bucky.{
   publish
 }
 import com.rabbitmq.client.LongString
+import com.typesafe.scalalogging.StrictLogging
 import dev.profunktor.fs2rabbit.arguments.SafeArg
 import dev.profunktor.fs2rabbit.config.Fs2RabbitConfig
 import dev.profunktor.fs2rabbit.config.declaration._
@@ -51,18 +52,20 @@ import dev.profunktor.fs2rabbit.model.{AMQPChannel, HeaderKey, Headers, Publishi
 import scodec.bits.ByteVector
 
 import java.util.{Date, UUID}
-import scala.concurrent.duration.FiniteDuration
+import scala.concurrent.duration._
 import scala.jdk.CollectionConverters._
 import scala.language.higherKinds
 import Fs2RabbitAmqpClient._
 import cats.effect.kernel.Temporal
 
 class Fs2RabbitAmqpClient[F[_]: Async: Temporal](
+    config: AmqpClientConfig,
     client: RabbitClient[F],
     connection: model.AMQPConnection,
     publishChannel: model.AMQPChannel,
     amqpClientConnectionManager: AmqpClientConnectionManager[F]
-) extends AmqpClient[F] {
+) extends AmqpClient[F]
+    with StrictLogging {
 
   override def declare(declarations: decl.Declaration*): F[Unit] = declare(declarations.toList)
 
@@ -181,37 +184,65 @@ class Fs2RabbitAmqpClient[F[_]: Async: Temporal](
       shutdownTimeout: FiniteDuration,
       shutdownRetry: FiniteDuration
   ): Resource[F, Unit] =
-    client.createChannel(connection).flatMap { implicit channel =>
-      implicit val decoder: EnvelopeDecoder[F, consume.Delivery] = deliveryDecoder(queueName)
-      Resource.eval(Ref.of[F, Set[UUID]](Set.empty)).flatMap { consumptionIds =>
-        Resource.eval(client.createAckerConsumer[consume.Delivery](model.QueueName(queueName.value))).flatMap { case (acker, consumer) =>
-          consumer
-            .evalMap(delivery =>
-              for {
-                uuid <- Async[F].delay(UUID.randomUUID())
-                _    <- consumptionIds.update(set => set + uuid)
-                res  <- handler(delivery.payload).attempt
-                tag = delivery.deliveryTag
-                _      <- consumptionIds.update(set => set - uuid)
-                result <- Async[F].fromEither(res)
-              } yield (result, tag)
-            )
-            .evalMap {
-              case (consume.Ack, tag)                => acker(model.AckResult.Ack(tag))
-              case (consume.DeadLetter, tag)         => acker(model.AckResult.NAck(tag))
-              case (consume.RequeueImmediately, tag) => acker(model.AckResult.Reject(tag))
-            }
-            .compile
-            .drain
-            .background
-            .flatMap { _ =>
-              Resource.onFinalize(
-                repeatUntil(consumptionIds.get)(_.isEmpty)(shutdownRetry).timeout(shutdownTimeout)
-              )
-            }
-            .map(_ => ())
+    Resource.eval(Ref.of[F, Set[UUID]](Set.empty)).flatMap { consumptionIds =>
+
+      // Create a fresh channel and consumer for each run. This ensures that after
+      // an auto-recovery the old (dead) fs2-rabbit stream is discarded and a new
+      // one is registered on the recovered channel rather than relying on the
+      // channel's internal queue which nobody is draining.
+      def runConsumer: F[Unit] =
+        client.createChannel(connection).use { implicit channel =>
+          implicit val decoder: EnvelopeDecoder[F, consume.Delivery] = deliveryDecoder(queueName)
+          client.createAckerConsumer[consume.Delivery](model.QueueName(queueName.value)).flatMap { case (acker, consumer) =>
+            consumer
+              .evalMap { delivery =>
+                val tag = delivery.deliveryTag
+
+                Async[F]
+                  .bracket {
+                    Async[F].delay(UUID.randomUUID()).flatTap(uuid => consumptionIds.update(_ + uuid))
+                  } { uuid =>
+                    handler(delivery.payload).attempt.flatMap {
+                      case Right(action) => Async[F].pure((action, tag))
+                      case Left(e) =>
+                        Async[F].delay(logger.error(s"Handler exception for queue ${queueName.value}: ${e.getMessage}", e)) *>
+                          Async[F].pure((exceptionalAction, tag))
+                    }
+                  } { uuid =>
+                    consumptionIds.update(_ - uuid)
+                  }
+              }
+              .evalMap {
+                case (consume.Ack, tag)                => acker(model.AckResult.Ack(tag))
+                case (consume.DeadLetter, tag)         => acker(model.AckResult.NAck(tag))
+                case (consume.RequeueImmediately, tag) => acker(model.AckResult.Reject(tag))
+              }
+              .compile
+              .drain
+          }
         }
-      }
+
+      // The delay between retry attempts. Aligned with the Java AMQP client's
+      // automatic-recovery interval so we wait long enough for the connection
+      // to be re-established before trying to open a new channel.
+      val recoveryDelay: FiniteDuration = config.networkRecoveryInterval.getOrElse(3.seconds)
+
+      // Retry the consumer indefinitely on failure.  This mirrors the
+      // AutorecoveringChannel behaviour of the v3 Java backend: when a network
+      // blip closes the channel, the consumer is re-registered automatically
+      // after the connection is recovered.
+      def consumerWithRecovery: F[Unit] =
+        runConsumer.handleErrorWith { error =>
+          Async[F].delay(logger.warn(s"Consumer for queue ${queueName.value} failed, will retry after $recoveryDelay: ${error.getMessage}")) *>
+            Temporal[F].sleep(recoveryDelay) *>
+            consumerWithRecovery
+        }
+
+      consumerWithRecovery.background.flatMap { _ =>
+        Resource.onFinalize(
+          repeatUntil(consumptionIds.get)(_.isEmpty)(shutdownRetry).timeout(shutdownTimeout)
+        )
+      }.map(_ => ())
     }
 
   override def isConnectionOpen: F[Boolean] = Async[F].pure(connection.value.isOpen)
@@ -246,7 +277,7 @@ object Fs2RabbitAmqpClient {
           amqpChannel = publishChannel
         )
       )
-    } yield new Fs2RabbitAmqpClient(client, connection, publishChannel, amqpClientConnectionManager)
+    } yield new Fs2RabbitAmqpClient(config, client, connection, publishChannel, amqpClientConnectionManager)
   }
 
   implicit def deliveryEncoder[F[_]: Async]: MessageEncoder[F, PublishCommand] =

backendFs2Rabbit/src/test/scala/fs2rabbit/ConsumerConnectionRecoverySpec.scala

@@ -0,0 +1,93 @@
+package fs2rabbit
+
+import cats.effect.IO
+import cats.effect.testing.scalatest.AsyncIOSpec
+import cats.implicits._
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AsyncWordSpec
+
+import java.io.IOException
+import scala.concurrent.duration._
+
+/** Regression test for the connection-recovery bug introduced in v4.
+  *
+  * v3 used `DefaultConsumer` on an `AutorecoveringChannel`, which the Java AMQP
+  * client transparently re-registers after a network blip.  v4's fs2-rabbit
+  * backend ran the consumer as an fs2 Stream in a `.background` fiber and ignored
+  * the fiber outcome.  When the stream terminated due to a connection drop the
+  * error was silently swallowed and no restart was attempted.
+  */
+class ConsumerConnectionRecoverySpec extends AsyncWordSpec with AsyncIOSpec with Matchers {
+
+  /** Retry `action` indefinitely on failure, sleeping `delay` between attempts.
+    * This is the pattern added to `Fs2RabbitAmqpClient.registerConsumer`.
+    */
+  private def consumerWithRecovery(action: IO[Unit], delay: FiniteDuration): IO[Unit] =
+    action.handleErrorWith { _ =>
+      IO.sleep(delay) *> consumerWithRecovery(action, delay)
+    }
+
+  "The fs2-rabbit consumer" when {
+    "the connection drops" should {
+
+      /* ------------------------------------------------------------------
+       * Demonstrates the BUG: the old .background pattern silently discards
+       * the failure and never restarts the consumer.
+       * ------------------------------------------------------------------ */
+      "NOT recover without a retry mechanism (demonstrates the bug)" in {
+        val failingConsumer: IO[Unit] =
+          IO.raiseError(new IOException("Connection reset by peer"))
+
+        // OLD (buggy) pattern: background + ignore outcome
+        failingConsumer.background
+          .use { _ => IO.sleep(100.millis) }
+          .flatMap { _ => IO.pure(succeed) }
+        // consumer died; in the real code nothing would ever be acked again
+      }
+
+      /* ------------------------------------------------------------------
+       * Demonstrates the FIX: handleErrorWith + recursive retry restarts
+       * the consumer after a failure, matching AutorecoveringChannel
+       * behaviour from v3.
+       * ------------------------------------------------------------------ */
+      "recover and resume processing after reconnection (demonstrates the fix)" in {
+        for {
+          messagesProcessed <- IO.ref(List.empty[Int])
+          attempt           <- IO.ref(0)
+
+          // First attempt fails (connection dropped); second attempt succeeds.
+          runConsumer: IO[Unit] = attempt.updateAndGet(_ + 1).flatMap {
+            case 1 => IO.raiseError(new IOException("Connection reset by peer"))
+            case _ => List(1, 2, 3, 4, 5).traverse_(i => messagesProcessed.update(_ :+ i))
+          }
+
+          _ <- consumerWithRecovery(runConsumer, delay = 50.millis)
+
+          messages <- messagesProcessed.get
+        } yield messages should have size 5
+      }
+
+      /* ------------------------------------------------------------------
+       * Verifies that handler exceptions use exceptionalAction rather than
+       * killing the consumer stream (secondary bug fixed alongside).
+       * ------------------------------------------------------------------ */
+      "not kill the consumer stream when a handler throws an exception" in {
+        for {
+          processedCount <- IO.ref(0)
+          attempt        <- IO.ref(0)
+
+          // First call simulates a handler exception; second call succeeds.
+          // With the fix the stream restarts and processes messages.
+          runConsumer: IO[Unit] = attempt.updateAndGet(_ + 1).flatMap {
+            case 1 => IO.raiseError(new RuntimeException("Handler blew up"))
+            case _ => processedCount.update(_ + 1)
+          }
+
+          _ <- consumerWithRecovery(runConsumer, delay = 50.millis)
+
+          count <- processedCount.get
+        } yield count shouldBe 1
+      }
+    }
+  }
+}