Bound clear and release by loadingTimeout

README.md

@@ -271,8 +271,9 @@ as well. It gets to retry, where before it would have hung.
 `Config.loadingTimeout`, failing their waiters with `ExpiredError`. The load itself is not
 cancelled, only detached from the cache. `loadingTimeout` defaults to the smaller of
 `expireAfterRead` and `expireAfterWrite`, so set it explicitly if the loads are legitimately slower
-than the expiration. It does not cover `clear` or the release of the cache: both wait for the loads
-in flight, and a load that never completes still hangs them.
+than the expiration. `clear` and the release of the cache wait for the loads in flight for at most
+`loadingTimeout` and then give up on them the same way, so a load that never completes no longer
+hangs them. A plain `Cache.loading` has no timeout and still waits.
 
 **Enumeration is weakly consistent.** `keys`, `values`, `values1`, `size`, `foldMap` and
 `foldMapPar` are served by the `ConcurrentHashMap` and no longer observe an atomic snapshot of the

scache/src/main/scala/com/evolution/scache/ExpiringCache.scala

@@ -247,7 +247,7 @@ object ExpiringCache {
     for {
       entryMap <- LoadingCache.EntryMap.of[F, K, TimestampedValue].toResource
       loadingSince <- Ref[F].of(Map.empty[K, (LoadingDeferred, Timestamp)]).toResource
-      cache <- LoadingCache.of(entryMap)
+      cache <- LoadingCache.of(entryMap, loadingTimeoutMs.millis.some)
       _ <- schedule(expireInterval) { removeExpiredAndCheckSize(entryMap, cache, loadingSince) }
       _ <- config
         .refresh
@@ -537,9 +537,8 @@ object ExpiringCache {
    *   smaller of `expireAfterRead` and `expireAfterWrite` is used. Note, that the load is not
    *   cancelled, only detached from the cache, and that this, too, is best effort: the eviction
    *   only happens on a cleanup run, so a load may outlive the timeout by up to one run interval.
-   *   It does not protect [[Cache#clear]] or the release of the cache: both unlink the entries
-   *   before awaiting the loads, out of reach of the eviction, so a load that never completes still
-   *   hangs them.
+   *   [[Cache#clear]] and the release of the cache wait for the loads in flight for at most this
+   *   long, counted from the `clear` itself, before giving up on them the same way.
    */
   final case class Config[F[_], -K, V](
     expireAfterRead: FiniteDuration,

scache/src/main/scala/com/evolution/scache/LoadingCache.scala

@@ -10,6 +10,7 @@ import com.evolution.scache.Cache.Directive
 import com.evolutiongaming.catshelper.ParallelHelper.*
 
 import java.util.concurrent.ConcurrentHashMap
+import scala.concurrent.duration.*
 import scala.jdk.CollectionConverters.*
 
 /**
@@ -56,7 +57,9 @@ import scala.jdk.CollectionConverters.*
  * Neither `remove` nor `clear` cancels a load in flight, so a load that outlives one of them still
  * has a value on its hands. After a `remove` it stores that value under the key again, putting the
  * key back into the cache; after a `clear` it stores it into the entry the `clear` has already
- * unlinked, and the `clear` is the one that awaits and releases it.
+ * unlinked, and the `clear` is the one that awaits and releases it, unless the load outstays the
+ * `loadingTimeout` of the `clear`, in which case the load is failed with [[ExpiredError]] and left
+ * to release its own value.
  *
  * `Removed` is a tombstone meaning "this `EntryRef` is no longer in the map, look the key up
  * again". It is needed because the two levels cannot be updated atomically together, so a fiber
@@ -83,6 +86,12 @@ import scala.jdk.CollectionConverters.*
  */
 private[scache] object LoadingCache {
 
+  /**
+   * How long `clear`, and hence the release of the cache, waits for a load in flight when no
+   * `loadingTimeout` is given explicitly.
+   */
+  val DefaultLoadingTimeout: FiniteDuration = 1.minute
+
   def of[F[_]: Async, K, V]: Resource[F, Cache[F, K, V]] = {
     for {
       entryMap <- EntryMap.of[F, K, V].toResource
@@ -93,12 +102,17 @@ private[scache] object LoadingCache {
   /**
    * Cache over an existing [[EntryMap]], clearing it, and thus releasing all the values, when the
    * resource is released.
+   *
+   * @param loadingTimeout
+   *   how long `clear`, and hence the release, waits for a load in flight before giving up on it,
+   *   see [[apply]].
    */
   def of[F[_]: Async, K, V](
     entryMap: EntryMap[F, K, V],
+    loadingTimeout: Option[FiniteDuration] = DefaultLoadingTimeout.some,
   ): Resource[F, Cache[F, K, V]] = {
     Resource.make {
-      apply(entryMap).pure[F]
+      apply(entryMap, loadingTimeout).pure[F]
     } { cache =>
       cache.clear.flatten
     }
@@ -195,9 +209,15 @@ private[scache] object LoadingCache {
   /**
    * Cache over an existing [[EntryMap]], which never releases the values it still holds, hence
    * meant to be wrapped into a resource by [[of]] rather than used directly.
+   *
+   * @param loadingTimeout
+   *   how long `clear` waits for a load in flight before giving up on it and failing the load, and
+   *   everyone waiting for it, with [[ExpiredError]]. `None` waits indefinitely, which makes a load
+   *   that never completes hang `clear` and cache's release.
    */
   def apply[F[_]: Async, K, V](
     entryMap: EntryMap[F, K, V],
+    loadingTimeout: Option[FiniteDuration] = DefaultLoadingTimeout.some,
   ): Cache[F, K, V] = {
 
     val F = Async[F]
@@ -1007,13 +1027,39 @@ private[scache] object LoadingCache {
        * the cache resource, an entry added while a large cache is being cleared can outlive the
        * cache itself, with its value never released.
        *
-       * Values of entries that are still loading are awaited before being released, which is why a
-       * load that never completes would make this, and the release of the cache resource, hang.
-       * `ExpiringCache.Config.loadingTimeout` does not help here: the entries are unlinked before
-       * being awaited, so the cleanup routine no longer sees them, and on release that routine is
-       * already stopped.
+       * Values of entries that are still loading are awaited before being released, for at most
+       * `loadingTimeout`:
+       *   - A load that is still running by then is given up on, its `deferred` is completed with
+       *     [[ExpiredError]], which fails the waiters and tells the loading fiber to release the
+       *     value it computes itself.
+       *   - Without a `loadingTimeout` a load that never completes makes this, and the release of
+       *     the cache resource, hang.
        */
       def clear: F[F[Unit]] = {
+
+        def awaitLoading(deferred: DeferredThrow[F, Entry[F, V]]): F[Option[Entry[F, V]]] = {
+          loadingTimeout.fold(deferred.getOption) { timeout =>
+            deferred
+              .getOption
+              .timeoutTo(
+                timeout,
+                deferred
+                  .complete(ExpiredError.asLeft)
+                  .productR { deferred.getOption },
+              )
+          }
+        }
+
+        def release(entryRef: EntryRef[F, V]): F[Unit] = {
+          entryRef
+            .get
+            .flatMap {
+              case state: EntryState.Value[F, V] => state.entry.release1
+              case state: EntryState.Loading[F, V] => awaitLoading(state.deferred).flatMap { _.foldMapM { _.release1 } }
+              case EntryState.Removed => ().pure[F]
+            }
+        }
+
         entryMap
           .keys
           .flatMap { keys =>
@@ -1024,12 +1070,7 @@ private[scache] object LoadingCache {
           }
           .flatMap { entryRefs =>
             entryRefs
-              .parFoldMap1 { entryRef =>
-                entryRef
-                  .getOption
-                  .flatMap { _.foldMapM { _.release1 } }
-                  .uncancelable
-              }
+              .parFoldMap1 { entryRef => release(entryRef).uncancelable }
               .start
           }
           .uncancelable

scache/src/test/scala/com/evolution/scache/ExpiringCacheSpec.scala

@@ -36,6 +36,14 @@ class ExpiringCacheSpec extends AsyncFunSuite with Matchers {
     `loading timeout does not expire loaded values`[IO].run()
   }
 
+  test("clear gives up on stuck loads") {
+    `clear gives up on stuck loads`[IO].run()
+  }
+
+  test("release gives up on stuck loads") {
+    `release gives up on stuck loads`[IO].run()
+  }
+
   test(s"refresh periodically") {
     refreshPeriodically[IO].run()
   }
@@ -181,6 +189,69 @@ class ExpiringCacheSpec extends AsyncFunSuite with Matchers {
     }
   }
 
+  private def `clear gives up on stuck loads`[F[_]: Async] = {
+    val config = ExpiringCache.Config[F, Int, Int](
+      expireAfterRead = 1.minute,
+      loadingTimeout = 100.millis.some,
+    )
+    ExpiringCache.of[F, Int, Int](config).use { cache =>
+      for {
+        started <- Deferred[F, Unit]
+        gate <- Deferred[F, Unit]
+        released <- Deferred[F, Unit]
+        loader <- cache
+          .getOrUpdate1(0) { started.complete(()) *> gate.get.as((0, 0, released.complete(()).void.some)) }
+          .attempt
+          .start
+        _ <- started.get
+        waiter <- cache.getOrUpdate(0) { 1.pure[F] }.attempt.start
+        result <- {
+          for {
+            _ <- Temporal[F].timeout(cache.clear.flatten, 2.seconds)
+            outcome <- Temporal[F].timeout(waiter.joinWithNever, 2.seconds)
+            _ <- Sync[F].delay { outcome should matchPattern { case Left(ExpiredError) => } }
+            _ <- gate.complete(())
+            outcome <- Temporal[F].timeout(loader.joinWithNever, 2.seconds)
+            _ <- Sync[F].delay { outcome should matchPattern { case Left(ExpiredError) => } }
+            _ <- Temporal[F].timeout(released.get, 2.seconds)
+            value <- cache.get(0)
+            _ <- Sync[F].delay { value shouldEqual none }
+          } yield {}
+        }.guarantee { gate.complete(()) *> loader.join.void }
+      } yield result
+    }
+  }
+
+  private def `release gives up on stuck loads`[F[_]: Async] = {
+    val config = ExpiringCache.Config[F, Int, Int](
+      expireAfterRead = 1.minute,
+      loadingTimeout = 100.millis.some,
+    )
+    for {
+      started <- Deferred[F, Unit]
+      gate <- Deferred[F, Unit]
+      released <- Deferred[F, Unit]
+      (cache, release) <- ExpiringCache.of[F, Int, Int](config).allocated
+      loader <- cache
+        .getOrUpdate1(0) { started.complete(()) *> gate.get.as((0, 0, released.complete(()).void.some)) }
+        .attempt
+        .start
+      _ <- started.get
+      // The release is uncancelable, hence run in a fiber of its own with the timeout on the join,
+      // so that a release that hangs fails the test rather than blocking it.
+      releasing <- release.start
+      result <- {
+        for {
+          _ <- Temporal[F].timeout(releasing.joinWithNever, 2.seconds)
+          _ <- gate.complete(())
+          outcome <- Temporal[F].timeout(loader.joinWithNever, 2.seconds)
+          _ <- Sync[F].delay { outcome should matchPattern { case Left(ExpiredError) => } }
+          _ <- Temporal[F].timeout(released.get, 2.seconds)
+        } yield {}
+      }.guarantee { gate.complete(()) *> loader.join.void *> releasing.join.void }
+    } yield result
+  }
+
   private def refreshPeriodically[F[_]: Async] = {
     val refresh = ExpiringCache.Refresh[Int](100.millis) { _.some.pure[F] }
     val config = ExpiringCache.Config(