Fix variable expiration overflowing with the maximum duration (fixes #…

…217)

When the duration is set to the maximum length, Long.MAX_VALUE nanoseconds,
the calcuation of expirationTime - currentTime > 0 may overflow and be
negative. This will not occur if the same thread calculates both timestamps.
It may occur across threads when the expirationTime is concurrently updated
using a later base time than t1's reading of the currentTime. This can
occur whenever the maintenance work is triggered to sweep expired entries
and a user thread accesses the entry. The later timestamp plus the maximum
duration results in an overflow, causing the remaining time to be negative,
and therefore causes the cache to expire the entry.

The internal maximum is now capped at Long.MAX_VALUE / 2 or ~150 years. This
should give a broad safety net to avoid these concurrency-inducing overflows
in normal code.

caffeine/src/main/java/com/github/benmanes/caffeine/cache/Async.java

@@ -15,6 +15,7 @@
  */
 package com.github.benmanes.caffeine.cache;
 
+import static com.github.benmanes.caffeine.cache.BoundedLocalCache.MAXIMUM_EXPIRY;
 import static java.util.Objects.requireNonNull;
 
 import java.io.Serializable;
@@ -31,7 +32,6 @@
  * @author [email protected] (Ben Manes)
  */
 final class Async {
-  static final long MAXIMUM_EXPIRY = (Long.MAX_VALUE >> 1); // 150 years
 
   private Async() {}
 
@@ -120,8 +120,7 @@ Object writeReplace() {
    * An expiry for asynchronous computations. When the value is being loaded this expiry returns
    * {@code Long.MAX_VALUE} to indicate that the entry should not be evicted due to an expiry
    * constraint. If the value is computed successfully the entry must be reinserted so that the
-   * expiration is updated and the expiration timeouts reflect the value once present. The value
-   * maximum range is reserved to coordinate the asynchronous life cycle.
+   * expiration is updated and the expiration timeouts reflect the value once present.
    */
   static final class AsyncExpiry<K, V> implements Expiry<K, CompletableFuture<V>>, Serializable {
     private static final long serialVersionUID = 1L;
@@ -134,33 +133,28 @@ static final class AsyncExpiry<K, V> implements Expiry<K, CompletableFuture<V>>,
 
     @Override
     public long expireAfterCreate(K key, CompletableFuture<V> future, long currentTime) {
-      if (isReady(future)) {
-        long duration = delegate.expireAfterCreate(key, future.join(), currentTime);
-        return Math.min(duration, MAXIMUM_EXPIRY);
-      }
-      return Long.MAX_VALUE;
+      return isReady(future)
+          ? delegate.expireAfterCreate(key, future.join(), currentTime)
+          : Long.MAX_VALUE;
     }
 
     @Override
     public long expireAfterUpdate(K key, CompletableFuture<V> future,
         long currentTime, long currentDuration) {
       if (isReady(future)) {
-        long duration = (currentDuration > MAXIMUM_EXPIRY)
+        return (currentDuration > MAXIMUM_EXPIRY)
             ? delegate.expireAfterCreate(key, future.join(), currentTime)
             : delegate.expireAfterUpdate(key, future.join(), currentTime, currentDuration);
-        return Math.min(duration, MAXIMUM_EXPIRY);
       }
       return Long.MAX_VALUE;
     }
 
     @Override
     public long expireAfterRead(K key, CompletableFuture<V> future,
         long currentTime, long currentDuration) {
-      if (isReady(future)) {
-        long duration = delegate.expireAfterRead(key, future.join(), currentTime, currentDuration);
-        return Math.min(duration, MAXIMUM_EXPIRY);
-      }
-      return Long.MAX_VALUE;
+      return isReady(future)
+          ? delegate.expireAfterRead(key, future.join(), currentTime, currentDuration)
+          : Long.MAX_VALUE;
     }
 
     Object writeReplace() {

caffeine/src/main/java/com/github/benmanes/caffeine/cache/BoundedLocalCache.java

@@ -154,6 +154,8 @@ abstract class BoundedLocalCache<K, V> extends BLCHeader.DrainStatusRef<K, V>
   static final double PERCENT_MAIN_PROTECTED = 0.80d;
   /** The maximum time window between entry updates before the expiration must be reordered. */
   static final long EXPIRE_WRITE_TOLERANCE = TimeUnit.SECONDS.toNanos(1);
+  /** The maximum duration before an entry expires. */
+  static final long MAXIMUM_EXPIRY = (Long.MAX_VALUE >> 1); // 150 years
 
   final ConcurrentHashMap<Object, Node<K, V>> data;
   @Nullable final CacheLoader<K, V> cacheLoader;
@@ -866,7 +868,7 @@ void refreshIfNeeded(Node<K, V> node, long now) {
     K key;
     V oldValue;
     long oldWriteTime = node.getWriteTime();
-    long refreshWriteTime = (now + Async.MAXIMUM_EXPIRY);
+    long refreshWriteTime = (now + MAXIMUM_EXPIRY);
     if (((now - oldWriteTime) > refreshAfterWriteNanos())
         && ((key = node.getKey()) != null) && ((oldValue = node.getValue()) != null)
         && node.casWriteTime(oldWriteTime, refreshWriteTime)) {
@@ -941,7 +943,7 @@ void refreshIfNeeded(Node<K, V> node, long now) {
   long expireAfterCreate(@Nullable K key, @Nullable V value, Expiry<K, V> expiry, long now) {
     if (expiresVariable() && (key != null) && (value != null)) {
       long duration = expiry.expireAfterCreate(key, value, now);
-      return (now + duration);
+      return (now + Math.min(duration, MAXIMUM_EXPIRY));
     }
     return 0L;
   }
@@ -961,7 +963,7 @@ long expireAfterUpdate(Node<K, V> node, @Nullable K key,
     if (expiresVariable() && (key != null) && (value != null)) {
       long currentDuration = Math.max(1, node.getVariableTime() - now);
       long duration = expiry.expireAfterUpdate(key, value, now, currentDuration);
-      return (now + duration);
+      return (now + Math.min(duration, MAXIMUM_EXPIRY));
     }
     return 0L;
   }
@@ -981,7 +983,7 @@ long expireAfterRead(Node<K, V> node, @Nullable K key,
     if (expiresVariable() && (key != null) && (value != null)) {
       long currentDuration = Math.max(1, node.getVariableTime() - now);
       long duration = expiry.expireAfterRead(key, value, now, currentDuration);
-      return (now + duration);
+      return (now + Math.min(duration, MAXIMUM_EXPIRY));
     }
     return 0L;
   }
@@ -3255,7 +3257,7 @@ final class BoundedVarExpiration implements VarExpiration<K, V> {
           long durationNanos = TimeUnit.NANOSECONDS.convert(duration, unit);
           synchronized (node) {
             now = cache.expirationTicker().read();
-            node.setVariableTime(now + durationNanos);
+            node.setVariableTime(now + Math.min(durationNanos, MAXIMUM_EXPIRY));
           }
           cache.afterRead(node, now, /* recordHit */ false);
         }

caffeine/src/test/java/com/github/benmanes/caffeine/cache/AsyncTest.java

@@ -15,7 +15,6 @@
  */
 package com.github.benmanes.caffeine.cache;
 
-import static com.github.benmanes.caffeine.cache.Async.MAXIMUM_EXPIRY;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.not;
@@ -138,17 +137,6 @@ public void asyncExpiry_completed() {
     verify(expiry.delegate).expireAfterRead(0, 100, 1L, 2L);
   }
 
-  @Test
-  public void asyncExpiry_bounded() {
-    AsyncExpiry<Integer, Integer> expiry = makeAsyncExpiry(
-        Long.MAX_VALUE, Long.MAX_VALUE, Long.MAX_VALUE);
-    CompletableFuture<Integer> future = CompletableFuture.completedFuture(100);
-
-    assertThat(expiry.expireAfterCreate(0, future, 1L), is(MAXIMUM_EXPIRY));
-    assertThat(expiry.expireAfterUpdate(0, future, 1L, 2L), is(MAXIMUM_EXPIRY));
-    assertThat(expiry.expireAfterRead(0, future, 1L, 2L), is(MAXIMUM_EXPIRY));
-  }
-
   @DataProvider(name = "successful")
   public Object[][] providesSuccessful() {
     return new Object[][] {{ CompletableFuture.completedFuture(1) }};

caffeine/src/test/java/com/github/benmanes/caffeine/cache/ExpireAfterVarTest.java

@@ -17,6 +17,7 @@
 
 import static com.github.benmanes.caffeine.cache.testing.CacheWriterVerifier.verifyWriter;
 import static com.github.benmanes.caffeine.cache.testing.HasRemovalNotifications.hasRemovalNotifications;
+import static com.github.benmanes.caffeine.testing.Awaits.await;
 import static com.github.benmanes.caffeine.testing.IsEmptyMap.emptyMap;
 import static com.github.benmanes.caffeine.testing.IsFutureValue.futureOf;
 import static java.util.concurrent.TimeUnit.NANOSECONDS;
@@ -27,6 +28,7 @@
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.not;
 import static org.hamcrest.Matchers.nullValue;
+import static org.hamcrest.Matchers.sameInstance;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.ArgumentMatchers.anyLong;
@@ -42,6 +44,7 @@
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.testng.annotations.Listeners;
 import org.testng.annotations.Test;
@@ -59,6 +62,7 @@
 import com.github.benmanes.caffeine.cache.testing.CacheValidationListener;
 import com.github.benmanes.caffeine.cache.testing.CheckNoStats;
 import com.github.benmanes.caffeine.cache.testing.CheckNoWriter;
+import com.github.benmanes.caffeine.testing.ConcurrentTestHarness;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Iterables;
@@ -72,6 +76,34 @@
 @Test(dataProviderClass = CacheProvider.class)
 public final class ExpireAfterVarTest {
 
+  @Test(dataProvider = "caches")
+  @CacheSpec(expiryTime = Expire.FOREVER,
+      expiry = { CacheExpiry.CREATE, CacheExpiry.WRITE, CacheExpiry.ACCESS })
+  public void expiry_bounds(Cache<Integer, Integer> cache, CacheContext context) {
+    context.ticker().advance(System.nanoTime());
+    AtomicBoolean running = new AtomicBoolean();
+    AtomicBoolean done = new AtomicBoolean();
+    Integer key = context.absentKey();
+    cache.put(key, key);
+
+    try {
+      ConcurrentTestHarness.execute(() -> {
+        while (!done.get()) {
+          context.ticker().advance(1, TimeUnit.MINUTES);
+          cache.get(key, Integer::new);
+          running.set(true);
+        }
+      });
+      await().untilTrue(running);
+      cache.cleanUp();
+      done.set(true);
+
+      assertThat(cache.get(key, Integer::new), sameInstance(key));
+    } finally {
+      done.set(true);
+    }
+  }
+
   /* ---------------- Create -------------- */
 
   @Test(dataProvider = "caches")

gradle/dependencies.gradle

@@ -59,7 +59,7 @@ ext {
     concurrentlinkedhashmap: '1.4.2',
     ehcache2: '2.10.4',
     ehcache3: '3.4.0',
-    elasticSearch: '6.2.1',
+    elasticSearch: '6.2.2',
     expiringMap: '0.5.8',
     jackrabbit: '1.8.2',
     jamm: '0.3.2',