Improve delay calculation for scheduled expiration

When the TimerWheel is used with the new `Scheduler` option, the delay to
the next expiration event has to be calculated. This was scanning wheels
and buckets sequentially (165 total) in the adjusted order and returning
the first time period found. This missed that a cascade operations may
occur, which could bring down an event that fires earlier. Thus the
bucket in the next wheel has to be peek'd at and the minimum delay
taken.

This doesn't change functionality, but merely improves the promptness
of the new active expiration feature (disabled by default).

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

@@ -247,16 +247,39 @@ public long getExpirationDelay() {
       for (int j = start; j < end; j++) {
         Node<K, V> sentinel = timerWheel[(j & mask)];
         Node<K, V> next = sentinel.getNextInVariableOrder();
-        if (sentinel != next) {
-          long buckets = (j - start);
-          long delay = (buckets << SHIFT[i]) - (nanos & spanMask);
-          return (delay > 0) ? delay : SPANS[i];
+        if (next == sentinel) {
+          continue;
         }
+        long buckets = (j - start);
+        long delay = (buckets << SHIFT[i]) - (nanos & spanMask);
+        delay = (delay > 0) ? delay : SPANS[i];
+
+        for (int k = i + 1; k < SHIFT.length; k++) {
+          long nextDelay = peekAhead(k);
+          delay = Math.min(delay, nextDelay);
+        }
+
+        return delay;
       }
     }
     return Long.MAX_VALUE;
   }
 
+  /**
+   * Returns the duration when the wheel's next bucket expires, or {@link Long.MAX_VALUE} if empty.
+   */
+  long peekAhead(int i) {
+    long ticks = (nanos >>> SHIFT[i]);
+    Node<K, V>[] timerWheel = wheel[i];
+
+    long spanMask = SPANS[i] - 1;
+    int mask = timerWheel.length - 1;
+    int probe = (int) ((ticks  + 1) & mask);
+    Node<K, V> sentinel = timerWheel[probe];
+    Node<K, V> next = sentinel.getNextInVariableOrder();
+    return (next == sentinel) ? Long.MAX_VALUE : (SPANS[i] - (nanos & spanMask));
+  }
+
   /**
    * Returns an unmodifiable snapshot map roughly ordered by the expiration time. The wheels are
    * evaluated in order, but the timers that fall within the bucket's range are not sorted. Beware

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

@@ -20,6 +20,7 @@
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.lessThan;
+import static org.hamcrest.Matchers.lessThanOrEqualTo;
 import static org.hamcrest.Matchers.not;
 import static org.hamcrest.Matchers.nullValue;
 import static org.mockito.ArgumentMatchers.any;
@@ -29,6 +30,7 @@
 import static org.mockito.Mockito.when;
 
 import java.lang.ref.ReferenceQueue;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Comparator;
@@ -120,8 +122,57 @@ public void schedule_fuzzy(long clock, long nanos, long[] times) {
     checkTimerWheel(nanos);
   }
 
+  @Test
+  public void getExpirationDelay_empty() {
+    when(cache.evictEntry(any(), any(), anyLong())).thenReturn(true);
+    timerWheel.nanos = NOW;
+
+    assertThat(timerWheel.getExpirationDelay(), is(Long.MAX_VALUE));
+  }
+
+  @Test
+  public void getExpirationDelay_firstWheel() {
+    when(cache.evictEntry(any(), any(), anyLong())).thenReturn(true);
+    timerWheel.nanos = NOW;
+
+    long delay = Duration.ofSeconds(1).toNanos();
+    timerWheel.schedule(new Timer(NOW + delay));
+    assertThat(timerWheel.getExpirationDelay(), is(lessThanOrEqualTo(SPANS[0])));
+  }
+
+  @Test
+  public void getExpirationDelay_lastWheel() {
+    when(cache.evictEntry(any(), any(), anyLong())).thenReturn(true);
+    timerWheel.nanos = NOW;
+
+    long delay = Duration.ofDays(14).toNanos();
+    timerWheel.schedule(new Timer(NOW + delay));
+    assertThat(timerWheel.getExpirationDelay(), is(lessThanOrEqualTo(delay)));
+  }
+
+  @Test
+  public void getExpirationDelay_hierarchy() {
+    when(cache.evictEntry(any(), any(), anyLong())).thenReturn(true);
+    timerWheel.nanos = NOW;
+
+    long t15 = NOW + Duration.ofSeconds(15).toNanos(); // in wheel[0]
+    long t80 = NOW + Duration.ofSeconds(80).toNanos(); // in wheel[1]
+    timerWheel.schedule(new Timer(t15));
+    timerWheel.schedule(new Timer(t80));
+
+    long t45 = NOW + Duration.ofSeconds(45).toNanos(); // discard T15, T80 in wheel[1]
+    timerWheel.advance(t45);
+
+    long t95 = NOW + Duration.ofSeconds(95).toNanos(); // in wheel[0], but expires after T80
+    timerWheel.schedule(new Timer(t95));
+
+    long expectedDelay = (t80 - t45);
+    long delay = timerWheel.getExpirationDelay();
+    assertThat(delay, is(lessThan(expectedDelay + SPANS[0]))); // cascaded T80 in wheel[1]
+  }
+
   @Test(dataProvider = "fuzzySchedule", invocationCount = 25)
-  public void getExpirationDelay(long clock, long nanos, long[] times) {
+  public void getExpirationDelay_fuzzy(long clock, long nanos, long[] times) {
     when(cache.evictEntry(any(), any(), anyLong())).thenReturn(true);
     timerWheel.nanos = clock;
     for (long timeout : times) {