From b53fe543cd8d05fcfb7bcd992f7e29b7e654f1e6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mareks=20Ramp=C4=81ns?= <8796159+mr-git@users.noreply.github.com> Date: Tue, 29 Sep 2026 18:43:25 +0300 Subject: [PATCH 1/9] add failing unit-test --- .../com/evolution/scache/SerialMapSpec.scala | 34 +++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/scache/src/test/scala/com/evolution/scache/SerialMapSpec.scala b/scache/src/test/scala/com/evolution/scache/SerialMapSpec.scala index 0040a68e..bc7f3027 100644 --- a/scache/src/test/scala/com/evolution/scache/SerialMapSpec.scala +++ b/scache/src/test/scala/com/evolution/scache/SerialMapSpec.scala @@ -8,6 +8,7 @@ import com.evolutiongaming.catshelper.SerialRef import org.scalatest.funsuite.AsyncFunSuite import org.scalatest.matchers.should.Matchers +import scala.concurrent.duration.* import scala.util.control.NoStackTrace class SerialMapSpec extends AsyncFunSuite with Matchers { @@ -61,6 +62,10 @@ class SerialMapSpec extends AsyncFunSuite with Matchers { `not leak on failures`[IO].run() } + test("not lose concurrent update when entry creator fails") { + `not lose concurrent update when entry creator fails`[IO].run() + } + test("modify serially for the same key") { `modify serially for the same key`[IO].run() } @@ -283,6 +288,35 @@ class SerialMapSpec extends AsyncFunSuite with Matchers { } } + private def `not lose concurrent update when entry creator fails`[F[_]: Async] = { + val key = "key" + for { + serialMap <- SerialMap.of[F, String, Int] + blocked <- Deferred[F, Unit] + acquired <- Deferred[F, Unit] + value0 = serialMap.modify(key) { _ => + for { + _ <- acquired.complete(()) + _ <- blocked.get + a <- TestError.raiseError[F, (Option[Int], Unit)] + } yield a + } + value0 <- value0.attempt.startEnsure + _ <- acquired.get + value1 <- serialMap.put(key, 1).startEnsure + // let `put` pick up the entry created by the failing `modify` and wait on its lock + _ <- Async[F].sleep(100.millis) + _ <- blocked.complete(()) + value0 <- value0.join + value1 <- value1.join + value2 <- serialMap.get(key) + } yield { + value0 shouldEqual Outcome.succeeded(IO.pure(TestError.asLeft)) + value1 shouldEqual Outcome.succeeded(IO.pure(none[Int])) + value2 shouldEqual 1.some + } + } + private def `modify serially for the same key`[F[_]: Async] = { val key = "key" for { From 5c82549995b5c6c647e8764dea92234c0489f167 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mareks=20Ramp=C4=81ns?= <8796159+mr-git@users.noreply.github.com> Date: Tue, 29 Sep 2026 19:33:53 +0300 Subject: [PATCH 2/9] add the fix --- .../com/evolution/scache/SerialMap.scala | 25 +++++++++++-------- 1 file changed, 15 insertions(+), 10 deletions(-) diff --git a/scache/src/main/scala/com/evolution/scache/SerialMap.scala b/scache/src/main/scala/com/evolution/scache/SerialMap.scala index 9003b4c2..83169c46 100644 --- a/scache/src/main/scala/com/evolution/scache/SerialMap.scala +++ b/scache/src/main/scala/com/evolution/scache/SerialMap.scala @@ -136,24 +136,29 @@ object SerialMap { self => def modify(state: State[V]) = { def onValue(value: Option[V]) = { - f(value).attempt.map { + f(value).attempt.flatMap[(State[V], F[A])] { case Right((Some(value), a)) => val state = State.full(value) val fa = a.pure[F] - (state, fa) + (state, fa).pure[F] case Right((None, a)) => - val state = State.removed + val state = State.removed[V] val fa = remove.as(a) - (state, fa) + (state, fa).pure[F] case Left(error) => - val fa = for { - added <- added.get - _ <- if (added) remove.void else ().pure[F] - a <- error.raiseError[F, A] - } yield a - (state, fa) + added.get.map { added => + if (added) { + // let [possible] waiting next caller retry with new value + val state = State.removed[V] + val fa = remove *> error.raiseError[F, A] + (state, fa) + } else { + val fa = error.raiseError[F, A] + (state, fa) + } + } } } From 2abee45ebe6d25da4572222b69743efb7e065bb6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mareks=20Ramp=C4=81ns?= <8796159+mr-git@users.noreply.github.com> Date: Tue, 29 Sep 2026 19:41:22 +0300 Subject: [PATCH 3/9] add cementing unit-test --- .../com/evolution/scache/SerialMapSpec.scala | 36 ++++++++++++++++++- 1 file changed, 35 insertions(+), 1 deletion(-) diff --git a/scache/src/test/scala/com/evolution/scache/SerialMapSpec.scala b/scache/src/test/scala/com/evolution/scache/SerialMapSpec.scala index bc7f3027..efb95597 100644 --- a/scache/src/test/scala/com/evolution/scache/SerialMapSpec.scala +++ b/scache/src/test/scala/com/evolution/scache/SerialMapSpec.scala @@ -66,6 +66,10 @@ class SerialMapSpec extends AsyncFunSuite with Matchers { `not lose concurrent update when entry creator fails`[IO].run() } + test("not lose concurrent update when modify of existing value fails") { + `not lose concurrent update when modify of existing value fails`[IO].run() + } + test("modify serially for the same key") { `modify serially for the same key`[IO].run() } @@ -304,7 +308,7 @@ class SerialMapSpec extends AsyncFunSuite with Matchers { value0 <- value0.attempt.startEnsure _ <- acquired.get value1 <- serialMap.put(key, 1).startEnsure - // let `put` pick up the entry created by the failing `modify` and wait on its lock + // make `put` lock on soon to fail `modify` on `value0` _ <- Async[F].sleep(100.millis) _ <- blocked.complete(()) value0 <- value0.join @@ -317,6 +321,36 @@ class SerialMapSpec extends AsyncFunSuite with Matchers { } } + private def `not lose concurrent update when modify of existing value fails`[F[_]: Async] = { + val key = "key" + for { + serialMap <- SerialMap.of[F, String, Int] + _ <- serialMap.put(key, 0) + blocked <- Deferred[F, Unit] + acquired <- Deferred[F, Unit] + value0 = serialMap.modify(key) { _ => + for { + _ <- acquired.complete(()) + _ <- blocked.get + a <- TestError.raiseError[F, (Option[Int], Unit)] + } yield a + } + value0 <- value0.attempt.startEnsure + _ <- acquired.get + value1 <- serialMap.put(key, 1).startEnsure + // make `put` lock on soon to fail `modify` on `value0` + _ <- Async[F].sleep(100.millis) + _ <- blocked.complete(()) + value0 <- value0.join + value1 <- value1.join + value2 <- serialMap.get(key) + } yield { + value0 shouldEqual Outcome.succeeded(IO.pure(TestError.asLeft)) + value1 shouldEqual Outcome.succeeded(IO.pure(0.some)) + value2 shouldEqual 1.some + } + } + private def `modify serially for the same key`[F[_]: Async] = { val key = "key" for { From e3073607bd0826eec050c99afcff898891e265a3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mareks=20Ramp=C4=81ns?= <8796159+mr-git@users.noreply.github.com> Date: Wed, 30 Sep 2026 11:42:24 +0300 Subject: [PATCH 4/9] address CodeRabbit's comment --- .../com/evolution/scache/SerialMap.scala | 11 +- .../com/evolution/scache/SerialMapSpec.scala | 106 +++++++++++++++++- 2 files changed, 112 insertions(+), 5 deletions(-) diff --git a/scache/src/main/scala/com/evolution/scache/SerialMap.scala b/scache/src/main/scala/com/evolution/scache/SerialMap.scala index 83169c46..17657f95 100644 --- a/scache/src/main/scala/com/evolution/scache/SerialMap.scala +++ b/scache/src/main/scala/com/evolution/scache/SerialMap.scala @@ -149,9 +149,9 @@ object SerialMap { self => case Left(error) => added.get.map { added => - if (added) { + if (added && state == State.Empty) { // let [possible] waiting next caller retry with new value - val state = State.removed[V] + val state = State.Removed val fa = remove *> error.raiseError[F, A] (state, fa) } else { @@ -183,8 +183,11 @@ object SerialMap { self => for { added <- Ref[F].of(false) - serialRef <- cache.getOrUpdate(key) { adding(added) } - a <- modify(serialRef, added).uncancelable + // uncancelable to not leak `State.Empty` entry + a <- cache + .getOrUpdate(key) { adding(added) } + .flatMap { serialRef => modify(serialRef, added) } + .uncancelable } yield a } diff --git a/scache/src/test/scala/com/evolution/scache/SerialMapSpec.scala b/scache/src/test/scala/com/evolution/scache/SerialMapSpec.scala index efb95597..15b38c85 100644 --- a/scache/src/test/scala/com/evolution/scache/SerialMapSpec.scala +++ b/scache/src/test/scala/com/evolution/scache/SerialMapSpec.scala @@ -1,6 +1,8 @@ package com.evolution.scache -import cats.effect.{Async, Concurrent, Deferred, IO, Outcome} +import cats.MonadThrow +import cats.effect.{Async, Concurrent, Deferred, IO, Outcome, Ref} +import cats.kernel.CommutativeMonoid import cats.syntax.all.* import com.evolution.scache.IOSuite.* import com.evolutiongaming.catshelper.CatsHelper.* @@ -70,6 +72,14 @@ class SerialMapSpec extends AsyncFunSuite with Matchers { `not lose concurrent update when modify of existing value fails`[IO].run() } + test("not lose concurrent put completed before entry creator acquired permit") { + `not lose concurrent put completed before entry creator acquired permit`.run() + } + + test("not leak entry when creator is canceled before acquiring permit") { + `not leak entry when creator is canceled before acquiring permit`.run() + } + test("modify serially for the same key") { `modify serially for the same key`[IO].run() } @@ -351,6 +361,54 @@ class SerialMapSpec extends AsyncFunSuite with Matchers { } } + private def `not lose concurrent put completed before entry creator acquired permit` = { + val key = "key" + Cache.loading[IO, String, SerialRef[IO, SerialMap.State[Int]]].use { cache => + for { + published <- Deferred[IO, Unit] + proceed <- Deferred[IO, Unit] + // pause the first caller between: setting value in cache AND it acquires the permit to modify it + pausing <- onFirstGetOrUpdate(cache) { published.complete(()) *> proceed.get.void } + serialMap = SerialMap(pausing) + value0 <- serialMap + .modify(key) { _ => TestError.raiseError[IO, (Option[Int], Unit)] } + .attempt + .start + _ <- published.get + value1 <- serialMap.put(key, 1) + _ <- proceed.complete(()) + value0 <- value0.joinWithNever + value2 <- serialMap.get(key) + } yield { + value0 shouldEqual TestError.asLeft + value1 shouldEqual none[Int] + value2 shouldEqual 1.some + } + } + } + + private def `not leak entry when creator is canceled before acquiring permit` = { + val key = "key" + Cache.loading[IO, String, SerialRef[IO, SerialMap.State[Int]]].use { cache => + for { + // cancel the first caller right after the entry is set in cache, before it acquires the permit to modify it + canceling <- onFirstGetOrUpdate(cache) { IO.canceled } + serialMap = SerialMap(canceling) + value0 <- serialMap + .modify(key) { _ => TestError.raiseError[IO, (Option[Int], Unit)] } + .start + value0 <- value0.join + keys <- serialMap.keys + size <- serialMap.size + } yield { + // creator is not interrupted, it proceeds to run `f` and to clean up on its failure + value0 shouldEqual Outcome.errored[IO, Throwable, Unit](TestError) + keys shouldEqual Set.empty + size shouldEqual 0 + } + } + } + private def `modify serially for the same key`[F[_]: Async] = { val key = "key" for { @@ -411,4 +469,50 @@ object SerialMapSpec { } case object TestError extends RuntimeException with NoStackTrace + + // run `hook` after the first `getOrUpdate` call returns, after the first entry has been stored in cache + def onFirstGetOrUpdate[K, V](cache: Cache[IO, K, V])(hook: IO[Unit]): IO[Cache[IO, K, V]] = { + Ref[IO].of(true).map { first => + new DelegatingCache(cache) { + override def getOrUpdate(key: K)(value: => IO[V]) = { + super.getOrUpdate(key)(value).flatTap { _ => + first.getAndSet(false).flatMap { first => hook.whenA(first) } + } + } + } + } + } + + class DelegatingCache[F[_]: MonadThrow, K, V](cache: Cache[F, K, V]) extends Cache.Abstract1[F, K, V] { + + def get(key: K) = cache.get(key) + + def get1(key: K) = cache.get1(key) + + def getOrUpdate(key: K)(value: => F[V]) = cache.getOrUpdate(key)(value) + + def getOrUpdate1[A](key: K)(value: => F[(A, V, Option[Release])]) = cache.getOrUpdate1(key)(value) + + def put(key: K, value: V, release: Option[Release]) = cache.put(key, value, release) + + def modify[A](key: K)(f: Option[V] => (A, Directive[F, V])) = cache.modify(key)(f) + + def contains(key: K) = cache.contains(key) + + def size = cache.size + + def keys = cache.keys + + def values = cache.values + + def values1 = cache.values1 + + def remove(key: K) = cache.remove(key) + + def clear: F[Released] = cache.clear + + def foldMap[A: CommutativeMonoid](f: (K, Either[F[V], V]) => F[A]) = cache.foldMap(f) + + def foldMapPar[A: CommutativeMonoid](f: (K, Either[F[V], V]) => F[A]) = cache.foldMapPar(f) + } } From fda6256339c7010fe5a08812ef50eb34654d8c89 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mareks=20Ramp=C4=81ns?= <8796159+mr-git@users.noreply.github.com> Date: Thu, 1 Oct 2026 16:20:52 +0300 Subject: [PATCH 5/9] add cases of slow-put-clear-put scenarios * `A` creates a `serialRef` (`key -> Empty`) and puts it in the map * `A` starts `f` to produce the value * `B` calls `cache.clear` and removes the `key -> Empty` mapping (`A` still keeps own reference to `serialRef`) * `B` (or something else) creates new entry (new `serialRef`) `key -> Empty` and puts it in the map * `B` executes `f` and updates entry in map to `key -> Full(B)` * `A` gets the failure from `f` and removes `key -> Full(B)` because `A` holds to `serialRef`, but `remove(key)` happens by `key` --- .../com/evolution/scache/SerialMapSpec.scala | 65 +++++++++++++++++++ 1 file changed, 65 insertions(+) diff --git a/scache/src/test/scala/com/evolution/scache/SerialMapSpec.scala b/scache/src/test/scala/com/evolution/scache/SerialMapSpec.scala index 15b38c85..74413461 100644 --- a/scache/src/test/scala/com/evolution/scache/SerialMapSpec.scala +++ b/scache/src/test/scala/com/evolution/scache/SerialMapSpec.scala @@ -76,6 +76,14 @@ class SerialMapSpec extends AsyncFunSuite with Matchers { `not lose concurrent put completed before entry creator acquired permit`.run() } + test("not lose put made after clear when entry creator fails") { + `not lose put made after clear when entry creator fails`[IO].run() + } + + test("not lose put made after clear when entry creator removes value") { + `not lose put made after clear when entry creator removes value`[IO].run() + } + test("not leak entry when creator is canceled before acquiring permit") { `not leak entry when creator is canceled before acquiring permit`.run() } @@ -387,6 +395,63 @@ class SerialMapSpec extends AsyncFunSuite with Matchers { } } + private def `not lose put made after clear when entry creator fails`[F[_]: Async] = { + val key = "key" + for { + serialMap <- SerialMap.of[F, String, Int] + blocked <- Deferred[F, Unit] + acquired <- Deferred[F, Unit] + value0 = serialMap.modify(key) { _ => + for { + _ <- acquired.complete(()) + _ <- blocked.get + a <- TestError.raiseError[F, (Option[Int], Unit)] + } yield a + } + value0 <- value0.attempt.startEnsure + _ <- acquired.get + // drop the `serialRef` of `value0` from the cache while `modify` is in not finished + _ <- serialMap.clear + // add new `serialRef` with value `Full(1)` + value1 <- serialMap.put(key, 1) + _ <- blocked.complete(()) + value0 <- value0.join + value2 <- serialMap.get(key) + } yield { + value0 shouldEqual Outcome.succeeded(IO.pure(TestError.asLeft)) + value1 shouldEqual none[Int] + value2 shouldEqual 1.some + } + } + + private def `not lose put made after clear when entry creator removes value`[F[_]: Async] = { + val key = "key" + for { + serialMap <- SerialMap.of[F, String, Int] + blocked <- Deferred[F, Unit] + acquired <- Deferred[F, Unit] + value0 = serialMap.modify(key) { _ => + for { + _ <- acquired.complete(()) + _ <- blocked.get + } yield (none[Int], ()) + } + value0 <- value0.startEnsure + _ <- acquired.get + // drop the `serialRef` of `value0` from the cache while `modify` is in not finished + _ <- serialMap.clear + // add new `serialRef` with value `Full(1)` + value1 <- serialMap.put(key, 1) + _ <- blocked.complete(()) + value0 <- value0.join + value2 <- serialMap.get(key) + } yield { + value0 shouldEqual Outcome.succeeded(IO.pure(())) + value1 shouldEqual none[Int] + value2 shouldEqual 1.some + } + } + private def `not leak entry when creator is canceled before acquiring permit` = { val key = "key" Cache.loading[IO, String, SerialRef[IO, SerialMap.State[Int]]].use { cache => From 3dce945a79bc778e01842558bc17ad19a028168a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mareks=20Ramp=C4=81ns?= <8796159+mr-git@users.noreply.github.com> Date: Thu, 1 Oct 2026 16:43:20 +0300 Subject: [PATCH 6/9] fix the slow-put-clear-put scenario --- .../main/scala/com/evolution/scache/SerialMap.scala | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/scache/src/main/scala/com/evolution/scache/SerialMap.scala b/scache/src/main/scala/com/evolution/scache/SerialMap.scala index 17657f95..4067646d 100644 --- a/scache/src/main/scala/com/evolution/scache/SerialMap.scala +++ b/scache/src/main/scala/com/evolution/scache/SerialMap.scala @@ -122,7 +122,12 @@ object SerialMap { self => def modify[A](key: K)(f: Option[V] => F[(Option[V], A)]) = { - def remove = cache.remove(key) + def remove(serialRef: SerialRef[F, State[V]]) = { + cache.modify(key) { + case Some(current) if current eq serialRef => ((), Cache.Directive.Remove) + case _ => ((), Cache.Directive.Ignore) + } + } def adding(added: Ref[F, Boolean]) = { for { @@ -144,7 +149,7 @@ object SerialMap { self => case Right((None, a)) => val state = State.removed[V] - val fa = remove.as(a) + val fa = remove(serialRef).as(a) (state, fa).pure[F] case Left(error) => @@ -152,7 +157,7 @@ object SerialMap { self => if (added && state == State.Empty) { // let [possible] waiting next caller retry with new value val state = State.Removed - val fa = remove *> error.raiseError[F, A] + val fa = remove(serialRef) *> error.raiseError[F, A] (state, fa) } else { val fa = error.raiseError[F, A] From 3fa5713f2994b54ce8b1896ea9ce998da336226e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mareks=20Ramp=C4=81ns?= <8796159+mr-git@users.noreply.github.com> Date: Thu, 1 Oct 2026 17:00:53 +0300 Subject: [PATCH 7/9] remove `added` usage --- .../com/evolution/scache/SerialMap.scala | 49 +++++++------------ 1 file changed, 18 insertions(+), 31 deletions(-) diff --git a/scache/src/main/scala/com/evolution/scache/SerialMap.scala b/scache/src/main/scala/com/evolution/scache/SerialMap.scala index 4067646d..6d8f8993 100644 --- a/scache/src/main/scala/com/evolution/scache/SerialMap.scala +++ b/scache/src/main/scala/com/evolution/scache/SerialMap.scala @@ -2,7 +2,7 @@ package com.evolution.scache import cats.Applicative import cats.effect.implicits.* -import cats.effect.{Concurrent, Ref} +import cats.effect.Concurrent import cats.syntax.all.* import com.evolutiongaming.catshelper.{Runtime, SerialRef} @@ -129,41 +129,31 @@ object SerialMap { self => } } - def adding(added: Ref[F, Boolean]) = { - for { - _ <- added.set(true) - serialRef <- SerialRef[F].of(State.empty[V]) - } yield serialRef - } - - def modify(serialRef: SerialRef[F, State[V]], added: Ref[F, Boolean]) = { + def modify(serialRef: SerialRef[F, State[V]]) = { def modify(state: State[V]) = { def onValue(value: Option[V]) = { - f(value).attempt.flatMap[(State[V], F[A])] { + f(value).attempt.map[(State[V], F[A])] { case Right((Some(value), a)) => val state = State.full(value) val fa = a.pure[F] - (state, fa).pure[F] + (state, fa) case Right((None, a)) => val state = State.removed[V] val fa = remove(serialRef).as(a) - (state, fa).pure[F] + (state, fa) + + case Left(error) if state == State.Empty => + // let [possible] waiting next caller retry with new value + val state = State.removed[V] + val fa = remove(serialRef) *> error.raiseError[F, A] + (state, fa) case Left(error) => - added.get.map { added => - if (added && state == State.Empty) { - // let [possible] waiting next caller retry with new value - val state = State.Removed - val fa = remove(serialRef) *> error.raiseError[F, A] - (state, fa) - } else { - val fa = error.raiseError[F, A] - (state, fa) - } - } + val fa = error.raiseError[F, A] + (state, fa) } } @@ -186,14 +176,11 @@ object SerialMap { self => } yield a } - for { - added <- Ref[F].of(false) - // uncancelable to not leak `State.Empty` entry - a <- cache - .getOrUpdate(key) { adding(added) } - .flatMap { serialRef => modify(serialRef, added) } - .uncancelable - } yield a + // uncancelable to not leak `State.Empty` entry + cache + .getOrUpdate(key) { SerialRef[F].of(State.empty[V]) } + .flatMap { serialRef => modify(serialRef) } + .uncancelable } def update[A](key: K)(f: Option[V] => F[Option[V]]) = { From 515ef89c2e27545f6c4d6f2a94b5013fe3ad55a4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mareks=20Ramp=C4=81ns?= <8796159+mr-git@users.noreply.github.com> Date: Thu, 1 Oct 2026 17:03:27 +0300 Subject: [PATCH 8/9] apply `fmt` --- scache/src/main/scala/com/evolution/scache/SerialMap.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scache/src/main/scala/com/evolution/scache/SerialMap.scala b/scache/src/main/scala/com/evolution/scache/SerialMap.scala index 6d8f8993..05c0a6f1 100644 --- a/scache/src/main/scala/com/evolution/scache/SerialMap.scala +++ b/scache/src/main/scala/com/evolution/scache/SerialMap.scala @@ -1,8 +1,8 @@ package com.evolution.scache import cats.Applicative -import cats.effect.implicits.* import cats.effect.Concurrent +import cats.effect.implicits.* import cats.syntax.all.* import com.evolutiongaming.catshelper.{Runtime, SerialRef} From 3ff6bf7f6146e157e71357e5afa6e9a6dd374716 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mareks=20Ramp=C4=81ns?= <8796159+mr-git@users.noreply.github.com> Date: Fri, 2 Oct 2026 18:10:46 +0300 Subject: [PATCH 9/9] address comment --- .../com/evolution/scache/SerialMap.scala | 3 +- .../com/evolution/scache/SerialMapSpec.scala | 38 +++++++++++++++++++ 2 files changed, 40 insertions(+), 1 deletion(-) diff --git a/scache/src/main/scala/com/evolution/scache/SerialMap.scala b/scache/src/main/scala/com/evolution/scache/SerialMap.scala index 05c0a6f1..23ec59ef 100644 --- a/scache/src/main/scala/com/evolution/scache/SerialMap.scala +++ b/scache/src/main/scala/com/evolution/scache/SerialMap.scala @@ -159,7 +159,8 @@ object SerialMap { self => def onRemoving = { val state = State.removed[V] - val fa = self.modify(key)(f) + // remove own `serialRef` - it might have not been removed, if it was still in `Loading` state + val fa = remove(serialRef) *> Concurrent[F].cede *> self.modify(key)(f) (state, fa).pure[F] } diff --git a/scache/src/test/scala/com/evolution/scache/SerialMapSpec.scala b/scache/src/test/scala/com/evolution/scache/SerialMapSpec.scala index 74413461..20d1ffe0 100644 --- a/scache/src/test/scala/com/evolution/scache/SerialMapSpec.scala +++ b/scache/src/test/scala/com/evolution/scache/SerialMapSpec.scala @@ -88,6 +88,10 @@ class SerialMapSpec extends AsyncFunSuite with Matchers { `not leak entry when creator is canceled before acquiring permit`.run() } + test("not get stuck on removed entry left in cache while it was loading") { + `not get stuck on removed entry left in cache while it was loading`.run() + } + test("modify serially for the same key") { `modify serially for the same key`[IO].run() } @@ -474,6 +478,27 @@ class SerialMapSpec extends AsyncFunSuite with Matchers { } } + private def `not get stuck on removed entry left in cache while it was loading` = { + val key = "key" + Cache.loading[IO, String, SerialRef[IO, SerialMap.State[Int]]].use { cache => + for { + // `LoadingCache.modify` passes `None` to `f` for an entry still in `Loading` state, + // which happens when a waiter wakes up between deferred completion and entry being set to `Value` + loading <- onFirstModify(cache) { _ => None } + serialMap = SerialMap(loading) + // the `serialRef` becomes `Removed`, but removal from cache is ignored as entry seems to be loading + value0 <- serialMap.modify(key) { _ => (none[Int], ()).pure[IO] } + value1 <- serialMap.put(key, 1).start + value1 <- value1.join.timeout(1.second) + value2 <- serialMap.get(key) + } yield { + value0 shouldEqual (()) + value1 shouldEqual Outcome.succeeded(IO.pure(none[Int])) + value2 shouldEqual 1.some + } + } + } + private def `modify serially for the same key`[F[_]: Async] = { val key = "key" for { @@ -548,6 +573,19 @@ object SerialMapSpec { } } + // pass value transformed by `hook` to `f` in the first `modify` call + def onFirstModify[K, V](cache: Cache[IO, K, V])(hook: Option[V] => Option[V]): IO[Cache[IO, K, V]] = { + Ref[IO].of(true).map { first => + new DelegatingCache(cache) { + override def modify[A](key: K)(f: Option[V] => (A, Cache.Directive[IO, V])) = { + first.getAndSet(false).flatMap { first => + super.modify(key) { value => f(if (first) hook(value) else value) } + } + } + } + } + } + class DelegatingCache[F[_]: MonadThrow, K, V](cache: Cache[F, K, V]) extends Cache.Abstract1[F, K, V] { def get(key: K) = cache.get(key)