Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Add a method to refresh only if lookup is a success, not an error. #159

Closed
wants to merge 6 commits into from
Closed

Add a method to refresh only if lookup is a success, not an error. #159

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
250a3ac
Add rollbackIfFailed parameter in refresh
kajebiii Sep 21, 2023
25c3933
Implement another operator `refreshValue`
kajebiii Sep 22, 2023
481871e
Return exit on promise
kajebiii Sep 22, 2023
3df9381
Implement test about `refreshValue`
kajebiii Sep 22, 2023
5da82d1
Add case for rollback to empty cache
kajebiii Sep 22, 2023
bc00941
Fix typo
kajebiii Sep 22, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 39 additions & 5 deletions zio-cache/shared/src/main/scala/zio/cache/Cache.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,12 @@ abstract class Cache[-Key, +Error, +Value] {
*/
def refresh(key: Key): IO[Error, Unit]

/**
* Computes the value associated with the specified key, with the lookup
* function, and puts it in the cache only if it is a value, not an error.
*/
def refreshValue(key: Key): IO[Error, Unit]

/**
* Invalidates the value associated with the specified key.
*/
Expand Down Expand Up @@ -247,7 +253,13 @@ object Cache {
}
}

override def refresh(in: In): IO[Error, Unit] =
def refresh(in: In): IO[Error, Unit] =
refresh(in, rollbackIfError = false)

def refreshValue(in: In): IO[Error, Unit] =
refresh(in, rollbackIfError = true)

private def refresh(in: In, rollbackIfError: Boolean): IO[Error, Unit] =
ZIO.suspendSucceedUnsafe { implicit u =>
val k = keyBy(in)
val promise = newPromise()
Expand All @@ -256,7 +268,8 @@ object Cache {
value = map.putIfAbsent(k, MapValue.Pending(new MapKey(k), promise))
}
val result = if (value eq null) {
lookupValueOf(in, promise)
val rollbackResultIfError = if (rollbackIfError) Right(None) else Left(())
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not sure this really makes sense. If the value isn't in the cache at all how is this any different from a get which will put the value in the cache even if it is an error? I think one of the issues here is we are treating refresh inconsistently with get.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I see your point. To be consistent with get we can say that we put the error into the cache then and remove that special case.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah I think our principle should be that the idea, as I understand it, is that if there is a valid "good" value in the cache we don't want to overwrite it with a "bad" value. One question that raises is what we should do when the original value is itself an error.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I do wonder whether this is the best path to address this. Like do you want to cache error values at all?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

  • Empty cache
    • Lookup Success -> replace with lookup exit
    • Lookup Error -> replace with lookup exit
  • Existing cache value
    • Lookup Success -> replace with lookup exit
    • Lookup Error
      • if the old one was a success, we keep it
      • if the old one was an error, do we keep it or replace with the new error? Replacing will reset the error ttl, so I think keeping the old one would be better since the main goal is to refresh the success value?

lookupValueOf(in, promise, rollbackResultIfError)
} else {
value match {
case MapValue.Pending(_, promiseInProgress) =>
Expand All @@ -267,7 +280,8 @@ object Cache {
get(in)
} else {
// Only trigger the lookup if we're still the current value, `completedResult`
lookupValueOf(in, promise).when {
val rollbackResultIfError = if (rollbackIfError) Right(Some(completedResult)) else Left(())
lookupValueOf(in, promise, rollbackResultIfError).when {
map.replace(k, completedResult, MapValue.Refreshing(promise, completedResult))
}
}
Expand All @@ -292,7 +306,16 @@ object Cache {
def size(implicit trace: Trace): UIO[Int] =
ZIO.succeed(map.size)

private def lookupValueOf(in: In, promise: Promise[Error, Value]): IO[Error, Value] =
private def lookupValueOf(
in: In,
promise: Promise[Error, Value],
/**
* Left(()): Put the lookup result.
* Right(None): Remove key if there is an error.
* Right(Some(rollbackResult)): Rollback if there is an error.
*/
rollbackResultIfError: Either[Unit, Option[MapValue.Complete[Key, Error, Value]]] = Left(())
): IO[Error, Value] =
ZIO.suspendSucceed {
val key = keyBy(in)
lookup(in)
Expand All @@ -302,7 +325,18 @@ object Cache {
val now = Unsafe.unsafe(implicit u => clock.unsafe.instant())
val entryStats = EntryStats(now)

map.put(key, MapValue.Complete(new MapKey(key), exit, entryStats, now.plus(timeToLive(exit))))
if (exit.isSuccess)
map.put(key, MapValue.Complete(new MapKey(key), exit, entryStats, now.plus(timeToLive(exit))))
else
rollbackResultIfError match {
case Left(()) =>
map.put(key, MapValue.Complete(new MapKey(key), exit, entryStats, now.plus(timeToLive(exit))))
case Right(None) =>
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Similarly here this is basically a get when there is no previous value.

map.remove(key)
case Right(Some(rollbackResult)) =>
map.put(key, rollbackResult)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Here if the time to live has already expired this is really another get.

}

promise.done(exit) *> ZIO.done(exit)
}
.onInterrupt(promise.interrupt *> ZIO.succeed(map.remove(key)))
Expand Down
93 changes: 91 additions & 2 deletions zio-cache/shared/src/test/scala/zio/cache/CacheSpec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -114,13 +114,17 @@ object CacheSpec extends ZIOSpecDefault {
_ <- cache.refresh(key)
val1 <- cache.get(key).either
_ <- cache.refresh(key)
val2 <- cache.get(key).either
failure2 <- cache.refresh(key).either
failure3 <- cache.get(key).either
_ <- cache.refresh(key)
val2 <- cache.get(key).either
val3 <- cache.get(key).either
} yield assert(failure1)(isLeft(equalTo(error))) &&
assert(failure2)(isLeft(equalTo(error))) &&
assert(failure3)(isLeft(equalTo(error))) &&
assert(val1)(isRight(equalTo(4))) &&
assert(val2)(isRight(equalTo(7)))
assert(val2)(isRight(equalTo(5))) &&
assert(val3)(isRight(equalTo(7)))
},
test("should get the value if the key doesn't exist in the cache") {
check(Gen.int) { salt =>
Expand All @@ -134,6 +138,91 @@ object CacheSpec extends ZIOSpecDefault {
}
}
),
suite("`refreshValue` method")(
test("should update the cache with a new value") {
def inc(n: Int) = n * 10

def retrieve(multiplier: Ref[Int])(key: Int) =
multiplier
.updateAndGet(inc)
.map(key * _)

val seed = 1
val key = 123
for {
ref <- Ref.make(seed)
cache <- Cache.make(1, Duration.Infinity, Lookup(retrieve(ref)))
val1 <- cache.get(key)
_ <- cache.refreshValue(key)
_ <- cache.get(key)
val2 <- cache.get(key)
} yield assertTrue(val1 == inc(key)) && assertTrue(val2 == inc(val1))
},
test("should update the cache only if lookup return a value, not an error.") {

val error = new RuntimeException("Must be a multiple of 3")

def inc(n: Int) = n + 1

def retrieve(number: Ref[Int])(key: Int) =
number
.updateAndGet(inc)
.flatMap {
case n if n % 3 == 0 =>
ZIO.fail(error)
case n =>
ZIO.succeed(key * n)
}

val seed = 2
val key = 1
for {
ref <- Ref.make(seed)
cache <- Cache.make(1, Duration.Infinity, Lookup(retrieve(ref)))
failure1 <- cache.get(key).either
_ <- cache.refreshValue(key)
val1 <- cache.get(key).either
_ <- cache.refreshValue(key)
val2 <- cache.get(key).either
failure2 <- cache.refreshValue(key).either
val3 <- cache.get(key).either
_ <- cache.refreshValue(key)
val4 <- cache.get(key).either
} yield assert(failure1)(isLeft(equalTo(error))) &&
assert(failure2)(isLeft(equalTo(error))) &&
assert(val1)(isRight(equalTo(4))) &&
assert(val2)(isRight(equalTo(5))) &&
assert(val3)(isRight(equalTo(5))) &&
assert(val4)(isRight(equalTo(7)))
},
test("should update only if it is a value when the key doesn't exist in the cache") {

val error = new RuntimeException("Must be a multiple of 3")

def inc(n: Int) = n + 1

def retrieve(number: Ref[Int])(key: Int) =
number
.updateAndGet(inc)
.flatMap {
case n if n % 3 == 0 =>
ZIO.fail(error)
case n =>
ZIO.succeed(key * n)
}

val seed = 2
val key = 1
val cap = 30
for {
ref <- Ref.make(seed)
cache <- Cache.make(cap, Duration.Infinity, Lookup(retrieve(ref)))
count0 <- cache.size
_ <- ZIO.foreachDiscard(1 to cap)(key => cache.refreshValue(key).either)
count1 <- cache.size
} yield assertTrue(count0 == 0) && assertTrue(count1 == cap / 3 * 2)
}
),
test("size") {
check(Gen.int) { salt =>
for {
Expand Down