Infinite Loop in MutableScatterMap::findFirstAvailableSlot?

[rsteppac]

Describe the bug
Our application uses two shared flows to process files. We use an instance of InMemoryKache to store filenames from both flows to avoid processing the same file twice. From time to time the application stops processing files altogether, while consuming 100% CPU. A thread dump shows that all threads are either parked or waiting with the exception of one thread with the following stacktrace:

      com.mayakapps.kache.collection.MutableScatterMap.findFirstAvailableSlot(ScatterMap.kt:989)
      com.mayakapps.kache.collection.MutableChainedScatterMap.resizeStorage$kache(MutableChainedScatterMap.kt:88)
      com.mayakapps.kache.collection.MutableScatterMap.adjustStorage$kache(ScatterMap.kt:1020)
      com.mayakapps.kache.collection.MutableScatterMap.findInsertIndex$kache(ScatterMap.kt:962)
      com.mayakapps.kache.collection.MutableScatterMap.put(ScatterMap.kt:752)
      com.mayakapps.kache.InMemoryKache.put(InMemoryKache.kt:225)
      com.our.app.scheduling.BaseUploadScheduler$uploadFiles$2$invokeSuspend$$inlined$filter$3$2.emit(Emitters.kt:223)
      com.our.app.scheduling.BaseUploadScheduler$uploadFiles$2$invokeSuspend$$inlined$filter$2$2.emit(Emitters.kt:219)
      com.our.app.scheduling.BaseUploadScheduler$uploadFiles$2$invokeSuspend$$inlined$filter$1$2.emit(Emitters.kt:219)
      com.our.app.scheduling.EventTriggerUploadScheduler$watchForNewFilesToUpload$$inlined$map$2$2.emit(Emitters.kt:219)
      com.our.app.scheduling.EventTriggerUploadScheduler$watchForNewFilesToUpload$$inlined$filter$1$2.emit(Emitters.kt:219)
      com.our.app.scheduling.EventTriggerUploadScheduler$watchForNewFilesToUpload$$inlined$map$1$2.emit(Emitters.kt:219)
      kotlinx.coroutines.flow.SharedFlowImpl.collect$suspendImpl(SharedFlow.kt:392)
      kotlinx.coroutines.flow.SharedFlowImpl$collect$1.invokeSuspend(SharedFlow.kt)
      kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:34)
      kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:104)
      io.opentelemetry.javaagent.instrumentation.kotlinxcoroutines.v1_0.RunnableWrapper.lambda$stopPropagation$0(RunnableWrapper.java:16)
      kotlinx.coroutines.scheduling.TaskImpl.run(Tasks.kt:99)
      kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely(CoroutineScheduler.kt:584)
      kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.executeTask(CoroutineScheduler.kt:811)
      kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.runWorker(CoroutineScheduler.kt:715)
      kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt:702)

To Reproduce
Concurrent access (put/remove) from two shared flows on a single instance of InMemoryKache.

Expected behavior
Concurrent access should not lead to infinite loop/stall the application.

System (please complete the following information):

• Device: PC
• OS: CentOS 7

Additional context

• Kotlin v2.3.20
• kache v2.1.1