class CachingTaskQueue {

  public synchronized SafeFuture<Optional<V>> perform(final CacheableTask<K, V> task) {
    // Check if a completed result is available
    final V cachedResult = cache.get(task.getKey());
    if (cachedResult != null) {
      cachedTaskCounter.inc();
      return SafeFuture.completedFuture(Optional.of(cachedResult));
    }

    // Check if the task is already scheduled
    final SafeFuture<Optional<V>> currentPendingTask = pendingTasks.get(task.getKey());
    if (currentPendingTask != null) {
      duplicateTaskCounter.inc();
      return currentPendingTask;
    }

    final SafeFuture<Optional<V>> generationResult = new SafeFuture<>();
    pendingTasks.put(task.getKey(), generationResult);

    // Check if there's a better starting point (in cache or in progress)
    final Optional<SafeFuture<Optional<V>>> newBase =
        task.streamIntermediateSteps()
            .map(
                key ->
                    Optional.ofNullable(cache.get(key))
                        .map(value -> SafeFuture.completedFuture(Optional.of(value)))
                        .orElse(pendingTasks.get(key)))
            .filter(Objects::nonNull)
            .findFirst();
    if (newBase.isPresent()) {
      rebasedTaskCounter.inc();
      newBase
          .get()
          .thenAccept(ancestorResult -> queueTask(task.rebase(ancestorResult)))
          .finish(error -> completePendingTask(task, Optional.empty(), error));
      return generationResult;
    }

    // Schedule the task for execution
    newTaskCounter.inc();
    queueTask(task);
    return generationResult;
  }

    CachingTaskQueue(
      final AsyncRunner asyncRunner,
      final MetricsSystem metricsSystem,
      final String metricsPrefix,
      final IntSupplier activeTaskLimit,
      final int maxCacheSize);

}