Issues with Scheduler

[aiguofer]

I've been debugging an issue where stale cache entries aren't getting evicted in low usage environments even when using a scheduler. After some debugging, I believe it's due to how Node.getAccessTime() gets calculated. I'm not exactly sure if the issue only happens when using aedile or whether it's also present in caffeine.

In order to debug, I created my own scheduler which logs out when a task is scheduled and adds a breakpoint to see where this gets called. Going through the debugger, I noticed sometimes the next scheduled event is FAAAR in the future because now - node.getAccessTime() sometimes returns a negative number in BoundedLocalCache.getExpirationDelay, so delay = Math.min(delay, expiresAfterAccessNanos() - (now - node.getAccessTime())); ends up bing very large.

Here's some logs from my scheduler:

For reference, here's my cache definition:

fun <T> createDataSourceCache(): Cache<DataSourceId, T> where T : Closeable, T : DataSource {
    return Caffeine.newBuilder()
        .expireAfterAccess(1.toDuration(DurationUnit.MINUTES))
        .scheduler(CustomCaffeineScheduler())
        .removalListener { key: DataSourceId?, value: T?, cause: RemovalCause ->
            logger.info { "Removing key='$key' from DataSource cache due to $cause" }
            try {
                value!!.close()
            } catch (e: Exception) {
                logger.error(e) { "Error removing data source from cache for key='$key'" }
            }
        }
        .asCache()
}

class CustomCaffeineScheduler : Scheduler {
    private val scheduler = Executors.newSingleThreadScheduledExecutor()

    override fun schedule(
        executor: Executor,
        command: Runnable,
        delay: Long,
        unit: TimeUnit,
    ): Future<*> {
        logger.info { "Scheduling $command with $delay $unit delay" }
        return scheduler.schedule({ executor.execute(command) }, delay, unit)
    }

    // Remember to shut down the scheduler when no longer needed
    fun shutdown() {
        scheduler.shutdown()
    }
}

Let me know if there's any other information I can provide!

[aiguofer]

Did a bit more digging and added some more logs.

Here's the method I use to retrieve the cache entry:

    /** Make sure only 1 thread can create a new datasource for a given environment. */
    private suspend fun lockedCacheGet(
        dataSourceId: DataSourceId,
        loader: suspend (DataSourceId) -> CloseableDataSource,
    ): CloseableDataSource {
        val mutex = environmentMutex.getOrPut(dataSourceId) { Mutex() }
        return mutex.withLock {
            logger.info { "[CAFFEINE] Attempting to retrieve entry $dataSourceId" }
            dataSourceCache.get(dataSourceId, loader)
        }
    }

And the loader method logs [CAFFEINE] Returning entry from loader function $dataSourceId.

Here's the log order:

It seems that the cleanup task gets scheduled after the initial attempt to get the value (this might be where that really large accessTime is coming from), but before the loader function is executed and the result is stored in the cache. However, when the result does get stored, it doesn't schedule a new task.

I have added the following workaround in my custom scheduler in the meantime:

    override fun schedule(
        executor: Executor,
        command: Runnable,
        delay: Long,
        unit: TimeUnit,
    ): Future<*> {
        logger.info {
            "[CAFFEINE] Scheduling $command with $delay $unit delay on executor $executor"
        }
        return scheduler.schedule(
            { executor.execute(command) },
            min(delay, CACHE_TTL.inWholeNanoseconds), // take the min to ensure this task always gets called within CACHE_TTL
            unit,
        )
    }