Rebuild head cache when a background task fails

journal/src/main/scala/com/evolution/kafka/journal/HeadCache.scala

@@ -161,8 +161,22 @@ object HeadCache {
           val log1 = log.prefixed(topic)
           cache
             .getOrUpdateResource(topic) {
+              val onBackgroundFailure = (e: Throwable) =>
+                log1.error(s"background task failed, rebuilding cache: $e", e) >>
+                  // Detach the eviction: `remove` releases this very resource (cancelling the
+                  // task that called us), so it must not run within this scope.
+                  cache.remove(topic).void.start.void
               TopicCache
-                .make(eventual, topic, log1, consumer, config, consRecordToActionHeader, metrics.map { _.headCache })
+                .make(
+                  eventual,
+                  topic,
+                  log1,
+                  consumer,
+                  config,
+                  consRecordToActionHeader,
+                  metrics.map { _.headCache },
+                  onBackgroundFailure,
+                )
                 .map { cache =>
                   metrics
                     .fold(cache) { metrics => cache.withMetrics(topic, metrics.headCache) }

journal/src/main/scala/com/evolution/kafka/journal/TopicCache.scala

@@ -70,12 +70,16 @@
    *   payload will be ignored.
    * @param metrics
    *   Interface to report the metrics to.
+   * @param onBackgroundFailure
+   *   Called once if any of the background tasks (Kafka consumption, Cassandra polling) terminates
+   *   abnormally. The intended use is to invalidate this cache so it gets rebuilt on the next
+   *   access, rather than silently degrading. It is not called on normal shutdown (cancellation).
    * @return
    *   Resource which will configure a [[TopicCache]] with the passed parameters. Instance of
    *   `Resource[TopicCache]` are, obviously, reusable and there is no need to call
    *   [[TopicCache#of]] each time if parameters did not change.
    */
   def make[F[_]: Async: Parallel](
     eventual: Eventual[F],
     topic: Topic,
     log: Log[F],
@@ -83,6 +87,7 @@
     config: HeadCacheConfig,
     consRecordToActionHeader: ConsRecordToActionHeader[F],
     metrics: Option[HeadCache.Metrics[F]],
+    onBackgroundFailure: Throwable => F[Unit],
   ): Resource[F, TopicCache[F]] = {
 
     for {
@@ -107,6 +112,19 @@
 
       cachesMap = caches.toMap
 
+      // Completed with the first background task failure. A background task is not expected to
+      // terminate on its own: consumption retries forever and the polling loops are `foreverM`, so
+      // any completion (error or unexpected success) means the cache stopped doing its job.
+      failure <- Deferred[F, Throwable].toResource
+      onBackgroundExit = (name: String) =>
+        (outcome: Outcome[F, Throwable, Unit]) =>
+          outcome match {
+            case Outcome.Succeeded(_) =>
+              failure.complete(JournalError(s"headcache background task '$name' terminated unexpectedly")).void
+            case Outcome.Errored(e) => failure.complete(e).void
+            case Outcome.Canceled() => ().pure[F]
+          }
+
       remove = caches
         .foldMapM {
           case (partition, cache) =>
@@ -170,7 +188,12 @@
                       cachesMap
                         .get(partition)
                         .fold {
-                          JournalError(s"invalid partition: $partition").raiseError[F, Sample]
+                          // A partition appeared that was not present when the cache was built
+                          // (the topic was repartitioned after init). `cachesMap` is fixed, so
+                          // retrying cannot recover: signal failure to rebuild the whole cache,
+                          // which re-reads partitions and picks up the new one.
+                          val error = JournalError(s"invalid partition: $partition")
+                          failure.complete(error) >> error.raiseError[F, Sample]
                         } { cache =>
                           cache
                             .add(records)
@@ -213,7 +236,8 @@
               }
           } yield result
         }
-        .onError { case e => log.error(s"consuming failed with $e", e) } /*TODO headcache: fail head cache*/
+        .onError { case e => log.error(s"consuming failed with $e", e) }
+        .guaranteeCase(onBackgroundExit("consuming"))
         .background
       random <- Random.State.fromClock[F]().toResource
       strategy = Strategy
@@ -226,7 +250,10 @@
         .retry(strategy)
         .handleErrorWith { e => log.error(s"remove failed, error: $e", e) }
         .foreverM[Unit]
+        .guaranteeCase(onBackgroundExit("remove"))
         .background
+      // Rebuild the cache when a background task dies, instead of serving stale/empty results forever.
+      _ <- failure.get.flatMap { e => onBackgroundFailure(e) }.background
       _ <- metrics.foldMapM { metrics =>
         val result = for {
           _ <- Temporal[F].sleep(1.minute)

journal/src/test/scala/com/evolution/kafka/journal/HeadCacheSpec.scala

@@ -25,6 +25,13 @@ import scala.util.control.NoStackTrace
 class HeadCacheSpec extends AsyncWordSpec with Matchers {
   import HeadCacheSpec.*
 
+  private def eventually[A](fa: IO[A], attempts: Int = 250, delay: FiniteDuration = 20.millis): IO[A] = {
+    fa.handleErrorWith { e =>
+      if (attempts <= 0) IO.raiseError(e)
+      else Temporal[IO].sleep(delay) >> eventually(fa, attempts - 1, delay)
+    }
+  }
+
   "HeadCache" should {
 
     "return result, records are in cache" in {
@@ -185,6 +192,52 @@ class HeadCacheSpec extends AsyncWordSpec with Matchers {
       result.run()
     }
 
+    "rebuild the cache when the topic is repartitioned" in {
+      val partition1 = Partition.unsafe(1)
+      val topicPartition1 = TopicPartition(topic = topic, partition = partition1)
+      val key0 = Key(id = "id0", topic = topic)
+      val key1 = Key(id = "id1", topic = topic)
+
+      def recordOn(tp: TopicPartition, key: Key): ConsumerRecords[String, Unit] = {
+        ConsumerRecordsOf(List(consumerRecordOf(appendOf(key, SeqNr.min), tp, Offset.min)))
+      }
+
+      def assignedPartitions(state: TestConsumer.State): Set[Partition] = {
+        state
+          .actions
+          .collect { case TestConsumer.Action.Assign(_, partitions) => partitions.toSortedSet }
+          .foldLeft(Set.empty[Partition]) { _ ++ _ }
+      }
+
+      val state = TestConsumer.State(topics = Map((topic, List(partition))))
+
+      val result = for {
+        stateRef <- Ref[IO].of(state)
+        consumer = TestConsumer.make(stateRef)
+        _ <- headCacheOf(HeadCache.Eventual.empty, consumer).use { headCache =>
+          for {
+            // build the per-topic cache while only partition 0 exists
+            _ <- stateRef.update { _.enqueue(recordOn(topicPartition, key0).pure[Try]) }
+            a <- headCache.get(key0, partition, Offset.min)
+            _ = a shouldEqual HeadInfo.append(Offset.min, SeqNr.min, none).some
+            // topic gets repartitioned: partition 1 now exists, and a record lands on it
+            _ <- stateRef.update { _.copy(topics = Map((topic, List(partition, partition1)))) }
+            _ <- stateRef.update { _.enqueue(recordOn(topicPartition1, key1).pure[Try]) }
+            // The unknown partition makes the cache fail and evict itself; the next `get` rebuilds
+            // it, re-reading partitions and this time assigning the new partition 1 (instead of
+            // looping forever). `get` drives the lazy rebuild; errors/None during recovery are
+            // tolerated until the assign is observed.
+            _ <- eventually {
+              headCache.get(key1, partition1, Offset.min).attempt >>
+                stateRef.get.map { s => assignedPartitions(s) should contain(partition1) }
+            }
+          } yield {}
+        }
+      } yield {}
+
+      result.run(10.seconds)
+    }
+
     "timeout" in {
       val headCache =
         headCacheOf(HeadCache.Eventual.empty, consumerEmpty.pure[IO].toResource, config.copy(timeout = 10.millis))