From 062f697ba017cc0dbb5353bcc6265b2077044a81 Mon Sep 17 00:00:00 2001 From: Alexander <12762063+appliedfunctor@users.noreply.github.com> Date: Fri, 26 Jun 2026 09:52:32 +0100 Subject: [PATCH 1/5] Fix consumer connection recovery in Fs2RabbitAmqpClient The v4 fs2-rabbit backend ran the consumer stream in a .background fiber and silently ignored the fiber outcome (.flatMap { _ => ... }). When a network blip caused the underlying channel to close, the stream terminated with an error that was swallowed and the consumer was never restarted. The v3 Java backend used DefaultConsumer on an AutorecoveringChannel; the Java AMQP client transparently re-registers those consumers after connection recovery. The fs2-rabbit backend had no equivalent mechanism. Two bugs are fixed: 1. Connection recovery: registerConsumer now wraps runConsumer in a recursive handleErrorWith retry loop (consumerWithRecovery) that sleeps for config.networkRecoveryInterval then creates a fresh channel and re-registers the consumer. A fresh channel is used on each retry to avoid the dead internal fs2-rabbit queue that the auto-recovered channel would otherwise accumulate messages into. 2. Handler exceptions: previously Async[F].fromEither(res) propagated handler errors through the stream and killed it. Now handler exceptions are logged and resolved via exceptionalAction, keeping the stream alive. Also adds config: AmqpClientConfig to Fs2RabbitAmqpClient so the recovery delay can be driven by networkRecoveryInterval, and StrictLogging for recovery/error logging. Regression tests added in ConsumerConnectionRecoverySpec. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .github/workflows/publish.yaml | 2 +- .../fs2rabbit/Fs2RabbitAmqpClient.scala | 97 ++++++++++++------- .../ConsumerConnectionRecoverySpec.scala | 93 ++++++++++++++++++ 3 files changed, 158 insertions(+), 34 deletions(-) create mode 100644 backendFs2Rabbit/src/test/scala/fs2rabbit/ConsumerConnectionRecoverySpec.scala diff --git a/.github/workflows/publish.yaml b/.github/workflows/publish.yaml index bce8328d..4195d022 100644 --- a/.github/workflows/publish.yaml +++ b/.github/workflows/publish.yaml @@ -49,7 +49,7 @@ jobs: - name: Publish if: ${{ startsWith(github.ref, 'refs/tags/') }} - run: sbt clean +publishSigned sonatypeCentralRelease + run: sbt clean +publishSigned sonatypeCentralUpload env: SONATYPE_USER: ${{ secrets.SONATYPE_USER }} SONATYPE_PASS: ${{ secrets.SONATYPE_PASS }} diff --git a/backendFs2Rabbit/src/main/scala/com/itv/bucky/backend/fs2rabbit/Fs2RabbitAmqpClient.scala b/backendFs2Rabbit/src/main/scala/com/itv/bucky/backend/fs2rabbit/Fs2RabbitAmqpClient.scala index b6321e12..09ebc652 100644 --- a/backendFs2Rabbit/src/main/scala/com/itv/bucky/backend/fs2rabbit/Fs2RabbitAmqpClient.scala +++ b/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 => + for { + uuid <- Async[F].delay(UUID.randomUUID()) + _ <- consumptionIds.update(_ + uuid) + res <- handler(delivery.payload).attempt + tag = delivery.deliveryTag + _ <- consumptionIds.update(_ - uuid) + // Use exceptionalAction for handler errors rather than propagating + // them through the stream (which would kill it unnecessarily). + result <- res match { + case Right(action) => Async[F].pure(action) + case Left(e) => + Async[F].delay(logger.error(s"Handler exception for queue ${queueName.value}: ${e.getMessage}", e)) *> + Async[F].pure(exceptionalAction) + } + } 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 + } } - } + + // 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] = diff --git a/backendFs2Rabbit/src/test/scala/fs2rabbit/ConsumerConnectionRecoverySpec.scala b/backendFs2Rabbit/src/test/scala/fs2rabbit/ConsumerConnectionRecoverySpec.scala new file mode 100644 index 00000000..5599b94d --- /dev/null +++ b/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 + } + } + } +} From 7b5dc1c25d52c022a4ea4935829067fba959d8f6 Mon Sep 17 00:00:00 2001 From: Alexander <12762063+appliedfunctor@users.noreply.github.com> Date: Fri, 26 Jun 2026 10:15:20 +0100 Subject: [PATCH 2/5] docs: document Fs2RabbitAmqpClient connection recovery fix in README Adds a '4.0.3 and above' section explaining the bug (consumer stream silently dying on connection drop with no restart), which backend is affected (Fs2RabbitAmqpClient only, not JavaBackendAmqpClient), how it was fixed, and directs users to upgrade to v4.0.3. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- README.md | 30 ++++++++++++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/README.md b/README.md index 67b809dc..279b8239 100644 --- a/README.md +++ b/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 From 6958c890aa18a05d4b5b485a2cdcf1f3471d0576 Mon Sep 17 00:00:00 2001 From: Alexander <12762063+appliedfunctor@users.noreply.github.com> Date: Fri, 26 Jun 2026 10:23:58 +0100 Subject: [PATCH 3/5] Revert unintended publish.yaml change Restores sonatypeCentralRelease (reverts accidental change to sonatypeCentralUpload that was included in the connection recovery commit). sonatypeCentralRelease uploads and automatically promotes the artifact; sonatypeCentralUpload only uploads and requires a manual release step. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .github/workflows/publish.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/publish.yaml b/.github/workflows/publish.yaml index 4195d022..bce8328d 100644 --- a/.github/workflows/publish.yaml +++ b/.github/workflows/publish.yaml @@ -49,7 +49,7 @@ jobs: - name: Publish if: ${{ startsWith(github.ref, 'refs/tags/') }} - run: sbt clean +publishSigned sonatypeCentralUpload + run: sbt clean +publishSigned sonatypeCentralRelease env: SONATYPE_USER: ${{ secrets.SONATYPE_USER }} SONATYPE_PASS: ${{ secrets.SONATYPE_PASS }} From 99fc914b25cdb897f903f7e835998b45b73aa60c Mon Sep 17 00:00:00 2001 From: Alexander Worton <12762063+appliedfunctor@users.noreply.github.com> Date: Fri, 26 Jun 2026 11:15:34 +0100 Subject: [PATCH 4/5] Cancellation-safe bracket for uuid removal Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- .../fs2rabbit/Fs2RabbitAmqpClient.scala | 28 +++++++++---------- 1 file changed, 14 insertions(+), 14 deletions(-) diff --git a/backendFs2Rabbit/src/main/scala/com/itv/bucky/backend/fs2rabbit/Fs2RabbitAmqpClient.scala b/backendFs2Rabbit/src/main/scala/com/itv/bucky/backend/fs2rabbit/Fs2RabbitAmqpClient.scala index 09ebc652..5d95b07e 100644 --- a/backendFs2Rabbit/src/main/scala/com/itv/bucky/backend/fs2rabbit/Fs2RabbitAmqpClient.scala +++ b/backendFs2Rabbit/src/main/scala/com/itv/bucky/backend/fs2rabbit/Fs2RabbitAmqpClient.scala @@ -196,21 +196,21 @@ class Fs2RabbitAmqpClient[F[_]: Async: Temporal]( client.createAckerConsumer[consume.Delivery](model.QueueName(queueName.value)).flatMap { case (acker, consumer) => consumer .evalMap { delivery => - for { - uuid <- Async[F].delay(UUID.randomUUID()) - _ <- consumptionIds.update(_ + uuid) - res <- handler(delivery.payload).attempt - tag = delivery.deliveryTag - _ <- consumptionIds.update(_ - uuid) - // Use exceptionalAction for handler errors rather than propagating - // them through the stream (which would kill it unnecessarily). - result <- res match { - case Right(action) => Async[F].pure(action) - case Left(e) => - Async[F].delay(logger.error(s"Handler exception for queue ${queueName.value}: ${e.getMessage}", e)) *> - Async[F].pure(exceptionalAction) + 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) } - } yield (result, tag) } .evalMap { case (consume.Ack, tag) => acker(model.AckResult.Ack(tag)) From 8e93b62dd1a49340e18dd3c497695dd9a477f1c6 Mon Sep 17 00:00:00 2001 From: Alexander <12762063+appliedfunctor@users.noreply.github.com> Date: Fri, 26 Jun 2026 14:18:53 +0100 Subject: [PATCH 5/5] Rewrite ConsumerConnectionRecoverySpec to exercise actual library code Extract acquireConsumerStream as a protected override point in Fs2RabbitAmqpClient so tests can inject controlled in-memory streams without a real RabbitMQ connection. Tests now subclass Fs2RabbitAmqpClient and override acquireConsumerStream to return a sequence of (acker, stream) pairs from a Ref, exercising the real registerConsumer implementation: - retry loop (consumerWithRecovery) recovers after a stream failure - handler exceptions are caught and routed to exceptionalAction rather than killing the stream - Ack/DeadLetter/RequeueImmediately map to the correct AckResult Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../fs2rabbit/Fs2RabbitAmqpClient.scala | 64 +++--- .../ConsumerConnectionRecoverySpec.scala | 194 ++++++++++++------ 2 files changed, 171 insertions(+), 87 deletions(-) diff --git a/backendFs2Rabbit/src/main/scala/com/itv/bucky/backend/fs2rabbit/Fs2RabbitAmqpClient.scala b/backendFs2Rabbit/src/main/scala/com/itv/bucky/backend/fs2rabbit/Fs2RabbitAmqpClient.scala index 5d95b07e..2042ab75 100644 --- a/backendFs2Rabbit/src/main/scala/com/itv/bucky/backend/fs2rabbit/Fs2RabbitAmqpClient.scala +++ b/backendFs2Rabbit/src/main/scala/com/itv/bucky/backend/fs2rabbit/Fs2RabbitAmqpClient.scala @@ -49,6 +49,7 @@ import dev.profunktor.fs2rabbit.model.AmqpFieldValue.{ TimestampVal } import dev.profunktor.fs2rabbit.model.{AMQPChannel, HeaderKey, Headers, PublishingFlag, ShortString} +import fs2.Stream import scodec.bits.ByteVector import java.util.{Date, UUID} @@ -176,6 +177,18 @@ class Fs2RabbitAmqpClient[F[_]: Async: Temporal]( _ <- if (ended) Async[F].unit else Temporal[F].sleep(sleep) *> repeatUntil(eval)(pred)(sleep) } yield () + /** Creates a fresh channel and registers a consumer on it. Extracted as a + * protected method so tests can override it with a controlled stream without + * needing a real RabbitMQ connection. + */ + protected def acquireConsumerStream( + queueName: bucky.QueueName + ): Resource[F, (model.AckResult => F[Unit], Stream[F, model.AmqpEnvelope[consume.Delivery]])] = + client.createChannel(connection).evalMap { implicit channel => + implicit val decoder: EnvelopeDecoder[F, consume.Delivery] = deliveryDecoder(queueName) + client.createAckerConsumer[consume.Delivery](model.QueueName(queueName.value)) + } + override def registerConsumer( queueName: bucky.QueueName, handler: Handler[F, consume.Delivery], @@ -191,35 +204,32 @@ class Fs2RabbitAmqpClient[F[_]: Async: Temporal]( // 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 + acquireConsumerStream(queueName).use { 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) + 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)) } - } - .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 - } + } { 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 diff --git a/backendFs2Rabbit/src/test/scala/fs2rabbit/ConsumerConnectionRecoverySpec.scala b/backendFs2Rabbit/src/test/scala/fs2rabbit/ConsumerConnectionRecoverySpec.scala index 5599b94d..44a418ec 100644 --- a/backendFs2Rabbit/src/test/scala/fs2rabbit/ConsumerConnectionRecoverySpec.scala +++ b/backendFs2Rabbit/src/test/scala/fs2rabbit/ConsumerConnectionRecoverySpec.scala @@ -1,92 +1,166 @@ package fs2rabbit -import cats.effect.IO +import cats.effect.{Deferred, IO, Ref, Resource} import cats.effect.testing.scalatest.AsyncIOSpec import cats.implicits._ +import com.itv.bucky.{AmqpClientConfig, Envelope, ExchangeName, Payload, QueueName, RoutingKey, consume, publish} +import com.itv.bucky.backend.fs2rabbit.{AmqpClientConnectionManager, Fs2RabbitAmqpClient} +import dev.profunktor.fs2rabbit.interpreter.RabbitClient +import dev.profunktor.fs2rabbit.model +import dev.profunktor.fs2rabbit.model.{AmqpEnvelope, AmqpProperties, DeliveryTag} +import fs2.Stream 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. +/** Regression tests 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 + * 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 + * the fiber outcome. When the stream terminated due to a connection drop the * error was silently swallowed and no restart was attempted. + * + * These tests exercise the actual `Fs2RabbitAmqpClient.registerConsumer` method + * by subclassing it and overriding `acquireConsumerStream` with controlled + * in-memory streams — no real RabbitMQ connection is required. */ 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`. + /** Build a minimal `AmqpEnvelope[consume.Delivery]` suitable for use in tests. */ + private def makeEnvelope(tag: Long): model.AmqpEnvelope[consume.Delivery] = + AmqpEnvelope( + DeliveryTag(tag), + consume.Delivery( + Payload("test".getBytes), + consume.ConsumerTag("ctag"), + Envelope(tag, redeliver = false, ExchangeName("ex"), RoutingKey("rk")), + publish.MessageProperties.minimalBasic + ), + AmqpProperties.empty, + model.ExchangeName("ex"), + model.RoutingKey("rk"), + redelivered = false + ) + + /** Config with a short recovery interval so tests don't wait 3 seconds. */ + private val testConfig: AmqpClientConfig = + AmqpClientConfig("localhost", 5672, "guest", "guest", networkRecoveryInterval = Some(50.millis)) + + /** Create a test double for `Fs2RabbitAmqpClient` that serves controlled + * `(acker, stream)` pairs from `streamsRef` in order. No real AMQP + * connection is needed: the overridden `acquireConsumerStream` never calls + * `client` or `connection`. */ - private def consumerWithRecovery(action: IO[Unit], delay: FiniteDuration): IO[Unit] = - action.handleErrorWith { _ => - IO.sleep(delay) *> consumerWithRecovery(action, delay) + private def makeTestClient( + streamsRef: Ref[IO, List[(model.AckResult => IO[Unit], Stream[IO, model.AmqpEnvelope[consume.Delivery]])]] + ): Fs2RabbitAmqpClient[IO] = + new Fs2RabbitAmqpClient[IO]( + testConfig, + null.asInstanceOf[RabbitClient[IO]], + null.asInstanceOf[model.AMQPConnection], + null.asInstanceOf[model.AMQPChannel], + null.asInstanceOf[AmqpClientConnectionManager[IO]] + ) { + override protected def acquireConsumerStream(queueName: QueueName) = + Resource.eval(streamsRef.modify { + case head :: tail => (tail, head) + case Nil => sys.error("No more test streams available") + }) } - "The fs2-rabbit consumer" when { - "the connection drops" should { + "Fs2RabbitAmqpClient.registerConsumer" when { - /* ------------------------------------------------------------------ - * 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")) + "the consumer stream fails (simulated connection drop)" should { - // 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 - } + "retry and resume processing messages after recovery" in { + // stream1 emits one message then fails, simulating a dropped connection + val stream1 = Stream.emit(makeEnvelope(1L)) ++ Stream.raiseError[IO](new IOException("Connection reset by peer")) + val acker1 = (_: model.AckResult) => IO.unit + // stream2 emits one more message then completes normally + val stream2 = Stream.emit(makeEnvelope(2L)) + val acker2 = (_: model.AckResult) => IO.unit - /* ------------------------------------------------------------------ - * 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 + processedTags <- IO.ref(List.empty[Long]) + done <- IO.deferred[Unit] + streamsRef <- IO.ref(List((acker1, stream1), (acker2, stream2))) + client = makeTestClient(streamsRef) + handler = (delivery: consume.Delivery) => + processedTags.update(_ :+ delivery.envelope.deliveryTag) *> + processedTags.get.flatMap(tags => if (tags.size >= 2) done.complete(()).void else IO.unit) *> + IO.pure(consume.Ack) + _ <- client + .registerConsumer(QueueName("test"), handler, consume.DeadLetter, 10, 500.millis, 100.millis) + .use(_ => done.get.timeout(5.seconds)) + tags <- processedTags.get + } yield { + tags should contain(1L) + tags should contain(2L) + } } + } - /* ------------------------------------------------------------------ - * 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) + "a handler throws an exception" should { - // 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) - } + "use exceptionalAction rather than killing the consumer stream" in { + for { + ackerResults <- IO.ref(List.empty[model.AckResult]) + done <- IO.deferred[Unit] + acker = (result: model.AckResult) => + ackerResults.update(_ :+ result) *> + ackerResults.get.flatMap(rs => if (rs.size >= 2) done.complete(()).void else IO.unit) + stream = Stream.emits(List(makeEnvelope(1L), makeEnvelope(2L))).covary[IO] + streamsRef <- IO.ref(List((acker, stream))) + client = makeTestClient(streamsRef) + // Handler throws for the first message; succeeds (Ack) for the second + handler = (delivery: consume.Delivery) => + if (delivery.envelope.deliveryTag == 1L) + IO.raiseError[consume.ConsumeAction](new RuntimeException("handler explosion")) + else + IO.pure(consume.Ack) + _ <- client + .registerConsumer(QueueName("test"), handler, consume.DeadLetter, 10, 500.millis, 100.millis) + .use(_ => done.get.timeout(5.seconds)) + results <- ackerResults.get + } yield { + // tag 1: handler threw → exceptionalAction (DeadLetter) → NAck + results should contain(model.AckResult.NAck(DeliveryTag(1L))) + // tag 2: handler returned Ack → Ack + results should contain(model.AckResult.Ack(DeliveryTag(2L))) + } + } + } - _ <- consumerWithRecovery(runConsumer, delay = 50.millis) + "messages are processed normally" should { - count <- processedCount.get - } yield count shouldBe 1 + "route Ack / DeadLetter / RequeueImmediately to the correct AckResult" in { + for { + ackerResults <- IO.ref(List.empty[model.AckResult]) + done <- IO.deferred[Unit] + acker = (result: model.AckResult) => + ackerResults.update(_ :+ result) *> + ackerResults.get.flatMap(rs => if (rs.size >= 3) done.complete(()).void else IO.unit) + stream = Stream.emits(List(makeEnvelope(1L), makeEnvelope(2L), makeEnvelope(3L))).covary[IO] + streamsRef <- IO.ref(List((acker, stream))) + client = makeTestClient(streamsRef) + handler = (delivery: consume.Delivery) => + delivery.envelope.deliveryTag match { + case 1L => IO.pure(consume.Ack) + case 2L => IO.pure(consume.DeadLetter) + case _ => IO.pure(consume.RequeueImmediately) + } + _ <- client + .registerConsumer(QueueName("test"), handler, consume.DeadLetter, 10, 500.millis, 100.millis) + .use(_ => done.get.timeout(5.seconds)) + results <- ackerResults.get + } yield { + results should contain(model.AckResult.Ack(DeliveryTag(1L))) + results should contain(model.AckResult.NAck(DeliveryTag(2L))) + results should contain(model.AckResult.Reject(DeliveryTag(3L))) + } } } }