Skip to content

ADFA-5331: Close JAR index sequences when indexing stops early #2126

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hal-eisen-adfa merged 10 commits into from
Oct 12, 2026
Merged

ADFA-5331: Close JAR index sequences when indexing stops early #2126

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
92689ff
style: spotless reformat, no functional change
hal-eisen-adfa Oct 11, 2026
5f8a511
ADFA-5331: Close JAR index sequences when indexing stops early
hal-eisen-adfa Oct 11, 2026
a0b8ad6
style: spotless reformat, no functional change
hal-eisen-adfa Oct 11, 2026
5758d6f
ADFA-5331: Address review findings F29, F30, F34, F36, F37, F38
hal-eisen-adfa Oct 11, 2026
2b3515a
ADFA-5331: Index JARs through a Flow so the JAR always closes (F31)
hal-eisen-adfa Oct 11, 2026
7959c1a
ADFA-5331: Untrack index jobs from a completion handler (F53)
hal-eisen-adfa Oct 11, 2026
9993278
style: spotless reformat, no functional change
hal-eisen-adfa Oct 12, 2026
5c6723b
ADFA-5331: Delete the unused scan() overloads in the sibling scanners
hal-eisen-adfa Oct 12, 2026
052c33a
ADFA-5331: State that partial rows are kept on any flow failure
hal-eisen-adfa Oct 12, 2026
d4a3cae
Merge branch 'stage' into bugfix/ADFA-5331
hal-eisen-adfa Oct 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
328 changes: 182 additions & 146 deletions lsp/indexing/src/main/kotlin/org/appdevforall/codeonthego/indexing/util/BackgroundIndexer.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,10 +6,12 @@ import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.cancelAndJoin
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.isActive
import kotlinx.coroutines.joinAll
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withContext
import org.appdevforall.codeonthego.indexing.api.Index
import org.appdevforall.codeonthego.indexing.api.Indexable
import org.slf4j.LoggerFactory
Expand All @@ -21,160 +23,194 @@ import java.util.concurrent.ConcurrentHashMap
* Implementations must be thread-safe.
*/
fun interface IndexingProgressListener {

/**
* Called with progress updates during indexing.
*
* @param sourceId The source being indexed.
* @param event What happened.
*/
fun onProgress(sourceId: String, event: IndexingEvent)
/**
* Called with progress updates during indexing.
*
* @param sourceId The source being indexed.
* @param event What happened.
*/
fun onProgress(
sourceId: String,
event: IndexingEvent,
)
}

sealed class IndexingEvent {
data object Started : IndexingEvent()
data class Progress(val processed: Int) : IndexingEvent()
data class Completed(val totalIndexed: Int) : IndexingEvent()
data class Failed(val error: Throwable) : IndexingEvent()
data object Skipped : IndexingEvent()
data object Started : IndexingEvent()

data class Progress(
val processed: Int,
) : IndexingEvent()

data class Completed(
val totalIndexed: Int,
) : IndexingEvent()

data class Failed(
val error: Throwable,
) : IndexingEvent()

data object Skipped : IndexingEvent()
}

/**
* Runs indexing operations in the background.
*/
class BackgroundIndexer<T : Indexable>(
private val index: Index<T>,
parentScope: CoroutineScope = CoroutineScope(Dispatchers.Default),
private val index: Index<T>,
parentScope: CoroutineScope = CoroutineScope(Dispatchers.Default),
) : Closeable {

private val job = SupervisorJob(parentScope.coroutineContext[Job])
private val scope = CoroutineScope(parentScope.coroutineContext + job)

companion object {
private val log = LoggerFactory.getLogger(BackgroundIndexer::class.java)
}

var progressListener: IndexingProgressListener? = null

private val activeJobs = ConcurrentHashMap<String, Job>()

/**
* Index a single source. The [provider] returns a [Sequence] that
* lazily produces entries — it is consumed on [Dispatchers.IO] by
* [Index.insertAll].
*
* If [skipIfExists] is true and the source is already indexed,
* this is a no-op.
*
* @param sourceId Identifies the source.
* @param skipIfExists Skip if already indexed.
* @param provider Lambda returning a [Sequence] of entries.
* @return The launched job.
*/
fun indexSource(
sourceId: String,
skipIfExists: Boolean = true,
provider: (sourceId: String) -> Sequence<T>,
): Job {
// Cancel any in-flight job for this source
activeJobs[sourceId]?.cancel()

val job = scope.launch {
try {
if (skipIfExists && index.containsSource(sourceId)) {
log.debug("Skipping already-indexed: {}", sourceId)
progressListener?.onProgress(sourceId, IndexingEvent.Skipped)
return@launch
}

log.info("Indexing: {}", sourceId)

// Remove stale entries first
index.removeBySource(sourceId)

if (!isActive) return@launch

progressListener?.onProgress(sourceId, IndexingEvent.Started)

var count = 0
val tracked = provider(sourceId).map { entry ->
count++
if (count % 1000 == 0) {
progressListener?.onProgress(sourceId, IndexingEvent.Progress(count))
}
entry
}

index.insertAll(tracked)

progressListener?.onProgress(sourceId, IndexingEvent.Completed(count))
log.info("Indexed {} entries from {}", count, sourceId)

} catch (e: CancellationException) {
log.debug("Indexing cancelled: {}", sourceId)
throw e
} catch (e: Exception) {
log.error("Indexing failed: {}", sourceId, e)
progressListener?.onProgress(sourceId, IndexingEvent.Failed(e))
} finally {
activeJobs.remove(sourceId)
}
}

activeJobs[sourceId] = job
return job
}

/**
* Index multiple sources sequentially in the background.
*
* Each source gets its own coroutine. The [SupervisorJob] ensures
* that one failure doesn't cancel the others.
*
* @param sources The sources to index (e.g. a list of JAR paths).
* @param mapper Maps each source to a (sourceId, Sequence) pair.
*/
fun <S> indexSources(
sources: Collection<S>,
skipIfExists: Boolean = true,
mapper: (S) -> Pair<String, Sequence<T>>,
): List<Job> {
return sources.map { source ->
val (sourceId, seq) = mapper(source)
indexSource(sourceId, skipIfExists) { seq }
}
}

/**
* Cancel all in-flight indexing and wait for completion.
*/
suspend fun cancelAll() {
activeJobs.values.toList().forEach { it.cancelAndJoin() }
}

/**
* Wait for all in-flight indexing to complete.
*/
suspend fun awaitAll() {
activeJobs.values.toList().joinAll()
}

/**
* Returns the number of currently active indexing jobs.
*/
val activeJobCount: Int get() = activeJobs.size

override fun close() {
val activeCount = activeJobCount
if (activeCount > 0) {
log.warn(
"Closing indexer with {} active job(s); cancellation is cooperative and close will wait for completion",
activeCount,
)
}
runBlocking {
job.cancelAndJoin()
}
activeJobs.clear()
}
private val job = SupervisorJob(parentScope.coroutineContext[Job])
private val scope = CoroutineScope(parentScope.coroutineContext + job)

companion object {
private val log = LoggerFactory.getLogger(BackgroundIndexer::class.java)
private const val BATCH_SIZE = 500
}

var progressListener: IndexingProgressListener? = null

private val activeJobs = ConcurrentHashMap<String, Job>()

/**
* Index a single source. The [provider] returns a [Flow] that is
* collected on [Dispatchers.IO] and inserted in batches through
* [Index.insertAll].
*
* If [skipIfExists] is true and the source is already indexed,
* this is a no-op.
*
* @param sourceId Identifies the source.
* @param skipIfExists Skip if already indexed.
* @param provider Lambda returning a [Flow] of entries. Resources the flow opens
* are released when collection completes, fails or is cancelled.
* @return The launched job.
*/
fun indexSource(
sourceId: String,
skipIfExists: Boolean = true,
provider: (sourceId: String) -> Flow<T>,
): Job {
// Cancel any in-flight job for this source
activeJobs[sourceId]?.cancel()

val job =
scope.launch {
try {
if (skipIfExists && index.containsSource(sourceId)) {
log.debug("Skipping already-indexed: {}", sourceId)
progressListener?.onProgress(sourceId, IndexingEvent.Skipped)
return@launch
}

log.info("Indexing: {}", sourceId)

// Remove stale entries first
index.removeBySource(sourceId)

if (!isActive) return@launch

progressListener?.onProgress(sourceId, IndexingEvent.Started)

var count = 0
withContext(Dispatchers.IO) {
val batch = ArrayList<T>(BATCH_SIZE)
var error: Exception? = null
try {
provider(sourceId).collect { entry ->
batch.add(entry)
count++
if (count % 1000 == 0) {
progressListener?.onProgress(sourceId, IndexingEvent.Progress(count))
}
if (batch.size == BATCH_SIZE) {
try {
index.insertAll(batch.asSequence())
} finally {
batch.clear()
}
}
}
} catch (e: CancellationException) {
throw e
} catch (e: Exception) {
// Keep the entries already collected on any failure, then report it. Meant for
// a failure after the last emit (e.g. EIO closing the JAR); a mid-stream failure
// also leaves a partial source that containsSource() treats as indexed.
error = e
}
if (batch.isNotEmpty()) index.insertAll(batch.asSequence())
Comment thread
hal-eisen-adfa marked this conversation as resolved.
error?.let { throw it }
Comment thread
hal-eisen-adfa marked this conversation as resolved.
}

progressListener?.onProgress(sourceId, IndexingEvent.Completed(count))
log.info("Indexed {} entries from {}", count, sourceId)
} catch (e: CancellationException) {
log.debug("Indexing cancelled: {}", sourceId)
throw e
} catch (e: Exception) {
log.error("Indexing failed: {}", sourceId, e)
progressListener?.onProgress(sourceId, IndexingEvent.Failed(e))
}
}

activeJobs[sourceId] = job
// Untrack on completion, not in the body: a job cancelled before it runs never reaches a
// finally, and a handler added after completion runs at once. Remove only this job: a
// replacement for the same source may already be registered.
job.invokeOnCompletion { activeJobs.remove(sourceId, job) }
return job
}

/**
* Index multiple sources sequentially in the background.
*
* Each source gets its own coroutine. The [SupervisorJob] ensures
* that one failure doesn't cancel the others.
*
* @param sources The sources to index (e.g. a list of JAR paths).
* @param mapper Maps each source to a (sourceId, Flow) pair.
*/
fun <S> indexSources(
sources: Collection<S>,
skipIfExists: Boolean = true,
mapper: (S) -> Pair<String, Flow<T>>,
): List<Job> =
sources.map { source ->
val (sourceId, flow) = mapper(source)
indexSource(sourceId, skipIfExists) { flow }
}

/**
* Cancel all in-flight indexing and wait for completion.
*/
suspend fun cancelAll() {
activeJobs.values.toList().forEach { it.cancelAndJoin() }
}

/**
* Wait for all in-flight indexing to complete.
*/
suspend fun awaitAll() {
activeJobs.values.toList().joinAll()
}

/**
* Returns the number of currently active indexing jobs.
*/
val activeJobCount: Int get() = activeJobs.size

override fun close() {
val activeCount = activeJobCount
if (activeCount > 0) {
log.warn(
"Closing indexer with {} active job(s); cancellation is cooperative and close will wait for completion",
activeCount,
)
}
runBlocking {
job.cancelAndJoin()
}
activeJobs.clear()
}
}
Loading
Loading