From 7750fef0746ea1cfbcdf6c25e01811c7a6b1ba2e Mon Sep 17 00:00:00 2001 From: Stas Shevchenko Date: Thu, 3 Sep 2026 17:39:37 +0200 Subject: [PATCH 1/2] Bound clear and release by loadingTimeout --- README.md | 5 +- .../com/evolution/scache/ExpiringCache.scala | 7 +- .../com/evolution/scache/LoadingCache.scala | 59 +++++++++++---- .../evolution/scache/ExpiringCacheSpec.scala | 71 +++++++++++++++++++ 4 files changed, 123 insertions(+), 19 deletions(-) diff --git a/README.md b/README.md index c6b24902..783d879c 100644 --- a/README.md +++ b/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 diff --git a/scache/src/main/scala/com/evolution/scache/ExpiringCache.scala b/scache/src/main/scala/com/evolution/scache/ExpiringCache.scala index de2f5a63..d8b559ef 100644 --- a/scache/src/main/scala/com/evolution/scache/ExpiringCache.scala +++ b/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, diff --git a/scache/src/main/scala/com/evolution/scache/LoadingCache.scala b/scache/src/main/scala/com/evolution/scache/LoadingCache.scala index bb98a0c5..d600ba1f 100644 --- a/scache/src/main/scala/com/evolution/scache/LoadingCache.scala +++ b/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.FiniteDuration 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 @@ -93,12 +96,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] = None, ): Resource[F, Cache[F, K, V]] = { Resource.make { - apply(entryMap).pure[F] + apply(entryMap, loadingTimeout).pure[F] } { cache => cache.clear.flatten } @@ -195,9 +203,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`. */ def apply[F[_]: Async, K, V]( entryMap: EntryMap[F, K, V], + loadingTimeout: Option[FiniteDuration] = None, ): Cache[F, K, V] = { val F = Async[F] @@ -1007,13 +1021,37 @@ 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`, if there is one. 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 +1062,7 @@ private[scache] object LoadingCache { } .flatMap { entryRefs => entryRefs - .parFoldMap1 { entryRef => - entryRef - .getOption - .flatMap { _.foldMapM { _.release1 } } - .uncancelable - } + .parFoldMap1 { entryRef => release(entryRef).uncancelable } .start } .uncancelable diff --git a/scache/src/test/scala/com/evolution/scache/ExpiringCacheSpec.scala b/scache/src/test/scala/com/evolution/scache/ExpiringCacheSpec.scala index 7067b0e3..6af2495f 100644 --- a/scache/src/test/scala/com/evolution/scache/ExpiringCacheSpec.scala +++ b/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( From b6bd261f827d7b824cc91c7041df6b99e778d534 Mon Sep 17 00:00:00 2001 From: Stas Shevchenko Date: Mon, 21 Sep 2026 20:06:37 +0200 Subject: [PATCH 2/2] Default loadingTimeout to one minute --- .../com/evolution/scache/LoadingCache.scala | 24 ++++++++++++------- 1 file changed, 16 insertions(+), 8 deletions(-) diff --git a/scache/src/main/scala/com/evolution/scache/LoadingCache.scala b/scache/src/main/scala/com/evolution/scache/LoadingCache.scala index d600ba1f..b0dca897 100644 --- a/scache/src/main/scala/com/evolution/scache/LoadingCache.scala +++ b/scache/src/main/scala/com/evolution/scache/LoadingCache.scala @@ -10,7 +10,7 @@ import com.evolution.scache.Cache.Directive import com.evolutiongaming.catshelper.ParallelHelper.* import java.util.concurrent.ConcurrentHashMap -import scala.concurrent.duration.FiniteDuration +import scala.concurrent.duration.* import scala.jdk.CollectionConverters.* /** @@ -86,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 @@ -103,7 +109,7 @@ private[scache] object LoadingCache { */ def of[F[_]: Async, K, V]( entryMap: EntryMap[F, K, V], - loadingTimeout: Option[FiniteDuration] = None, + loadingTimeout: Option[FiniteDuration] = DefaultLoadingTimeout.some, ): Resource[F, Cache[F, K, V]] = { Resource.make { apply(entryMap, loadingTimeout).pure[F] @@ -207,11 +213,11 @@ private[scache] object LoadingCache { * @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`. + * that never completes hang `clear` and cache's release. */ def apply[F[_]: Async, K, V]( entryMap: EntryMap[F, K, V], - loadingTimeout: Option[FiniteDuration] = None, + loadingTimeout: Option[FiniteDuration] = DefaultLoadingTimeout.some, ): Cache[F, K, V] = { val F = Async[F] @@ -1022,10 +1028,12 @@ private[scache] object LoadingCache { * cache itself, with its value never released. * * Values of entries that are still loading are awaited before being released, for at most - * `loadingTimeout`, if there is one. 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. + * `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]] = {