Implement proximity grouping for indexers based on block numbers and … (#101)

* Implement proximity grouping for indexers based on block numbers and dependencies

* Enhance proximity grouping logging with detailed summary of groups and indexers

* Refactor test for cross-group dependency extraction to clarify standalone group behavior

Author: daithihearn

src/main/kotlin/org.vechain.indexer/IndexerRunner.kt

@@ -1,5 +1,7 @@
 package org.vechain.indexer
 
+import kotlin.time.Duration
+import kotlin.time.Duration.Companion.minutes
 import kotlin.time.TimeMark
 import kotlin.time.TimeSource
 import kotlinx.coroutines.CoroutineScope
@@ -17,6 +19,7 @@ import org.vechain.indexer.exception.ReorgException
 import org.vechain.indexer.thor.client.ThorClient
 import org.vechain.indexer.utils.ClauseIndexMapping
 import org.vechain.indexer.utils.ClauseUtils.buildClauseListWithMapping
+import org.vechain.indexer.utils.IndexerOrderUtils.proximityGroups
 import org.vechain.indexer.utils.IndexerOrderUtils.topologicalOrder
 import org.vechain.indexer.utils.retryOnFailure
 
@@ -28,7 +31,9 @@ class IndexerRunner(private val timeSource: TimeSource = TimeSource.Monotonic) {
             scope: CoroutineScope,
             thorClient: ThorClient,
             indexers: List<Indexer>,
-            blockBatchSize: Int = 1
+            blockBatchSize: Int = 1,
+            proximityThreshold: Long = 1_000_000L,
+            reshuffleInterval: Duration = 5.minutes,
         ): Job {
             require(indexers.isNotEmpty()) { "At least one indexer is required" }
 
@@ -39,6 +44,8 @@ class IndexerRunner(private val timeSource: TimeSource = TimeSource.Monotonic) {
                     indexers = indexers,
                     batchSize = blockBatchSize,
                     thorClient = thorClient,
+                    proximityThreshold = proximityThreshold,
+                    reshuffleInterval = reshuffleInterval,
                 )
             }
         }
@@ -48,6 +55,8 @@ class IndexerRunner(private val timeSource: TimeSource = TimeSource.Monotonic) {
         indexers: List<Indexer>,
         batchSize: Int,
         thorClient: ThorClient,
+        proximityThreshold: Long = 1_000_000L,
+        reshuffleInterval: Duration = 5.minutes,
     ): Unit = coroutineScope {
         require(indexers.isNotEmpty()) { "At least one indexer is required" }
 
@@ -63,11 +72,19 @@ class IndexerRunner(private val timeSource: TimeSource = TimeSource.Monotonic) {
                     nonFastSyncable,
                     thorClient,
                     batchSize,
+                    proximityThreshold,
+                    reshuffleInterval,
                 )
                 // Re-initialise non-fast-syncable indexers to recover from potential
                 // mid-block cancellation during the intermediate run
                 initialise(nonFastSyncable)
-                runIndexers(indexers, thorClient, batchSize)
+                runWithProximityGroups(
+                    indexers,
+                    thorClient,
+                    batchSize,
+                    proximityThreshold,
+                    reshuffleInterval,
+                )
             } catch (e: ReorgException) {
                 logger.error("Reorg detected, restarting all indexers", e)
                 // Exception caught, job will complete normally and loop will restart
@@ -85,6 +102,8 @@ class IndexerRunner(private val timeSource: TimeSource = TimeSource.Monotonic) {
         nonFastSyncable: List<Indexer>,
         thorClient: ThorClient,
         batchSize: Int,
+        proximityThreshold: Long = 1_000_000L,
+        reshuffleInterval: Duration = 5.minutes,
     ) {
         if (fastSyncable.isEmpty()) {
             initialise(nonFastSyncable)
@@ -100,7 +119,15 @@ class IndexerRunner(private val timeSource: TimeSource = TimeSource.Monotonic) {
             val nonFastJob: Job? =
                 if (independentNonFast.isNotEmpty()) {
                     initialise(independentNonFast)
-                    launch { runIndexers(independentNonFast, thorClient, batchSize) }
+                    launch {
+                        runWithProximityGroups(
+                            independentNonFast,
+                            thorClient,
+                            batchSize,
+                            proximityThreshold,
+                            reshuffleInterval,
+                        )
+                    }
                 } else null
 
             initialiseAndSync(fastSyncable)
@@ -149,6 +176,44 @@ class IndexerRunner(private val timeSource: TimeSource = TimeSource.Monotonic) {
         }
     }
 
+    suspend fun runWithProximityGroups(
+        indexers: List<Indexer>,
+        thorClient: ThorClient,
+        batchSize: Int,
+        proximityThreshold: Long,
+        reshuffleInterval: Duration,
+    ) {
+        while (true) {
+            val groups = proximityGroups(indexers, proximityThreshold)
+            if (groups.size <= 1) {
+                // Steady state — single group, no deadline
+                runIndexers(indexers, thorClient, batchSize)
+                return
+            }
+
+            val groupSummary = buildString {
+                appendLine(
+                    "Proximity groups: ${groups.size} groups, ${indexers.size} indexers, threshold=$proximityThreshold"
+                )
+                groups.forEachIndexed { i, g ->
+                    val blockRange =
+                        "${g.minOf { it.getCurrentBlockNumber() }}..${g.maxOf { it.getCurrentBlockNumber() }}"
+                    appendLine(
+                        "  Group ${i + 1} (${g.size} indexers, blocks $blockRange): ${g.map { it.name }}"
+                    )
+                }
+            }
+            logger.info(groupSummary.trimEnd())
+            val deadlineMark = timeSource.markNow() + reshuffleInterval
+            coroutineScope {
+                groups.forEach { group ->
+                    launch { runIndexers(group, thorClient, batchSize, deadlineMark) }
+                }
+            }
+            // All groups completed naturally when deadline passed; loop to reshuffle
+        }
+    }
+
     suspend fun runIndexers(
         indexers: List<Indexer>,
         thorClient: ThorClient,

src/main/kotlin/org.vechain.indexer/utils/IndexerOrderUtils.kt

@@ -57,6 +57,99 @@ internal object IndexerOrderUtils {
         return listOf(ordered)
     }
 
+    /**
+     * Groups indexers by proximity of their current block numbers, preserving dependency chains.
+     *
+     * Indexers within [threshold] blocks of each other share a group. Cross-group dependency chains
+     * are extracted and re-merged into the closest group (or kept standalone). Each group is
+     * topologically ordered internally.
+     */
+    fun proximityGroups(indexers: List<Indexer>, threshold: Long): List<List<Indexer>> {
+        if (indexers.isEmpty()) return emptyList()
+
+        // Phase A: Initial proximity grouping by current block number
+        val sorted = indexers.sortedBy { it.getCurrentBlockNumber() }
+        val groups = mutableListOf<MutableSet<Indexer>>()
+        var currentGroup = mutableSetOf(sorted.first())
+        for (i in 1 ..< sorted.size) {
+            val gap = sorted[i].getCurrentBlockNumber() - sorted[i - 1].getCurrentBlockNumber()
+            if (gap > threshold) {
+                groups.add(currentGroup)
+                currentGroup = mutableSetOf()
+            }
+            currentGroup.add(sorted[i])
+        }
+        groups.add(currentGroup)
+
+        // Phase B: Extract cross-group dependency chains
+        val indexerSet = indexers.toSet()
+        val indexerToGroup = mutableMapOf<Indexer, MutableSet<Indexer>>()
+        for (group in groups) {
+            for (indexer in group) {
+                indexerToGroup[indexer] = group
+            }
+        }
+
+        val extracted = mutableListOf<MutableSet<Indexer>>()
+        val alreadyExtracted = mutableSetOf<Indexer>()
+
+        for (indexer in indexers) {
+            val dep = indexer.dependsOn ?: continue
+            val indexerGroup = indexerToGroup[indexer] ?: continue
+            val depGroup = indexerToGroup[dep] ?: continue
+            if (indexerGroup === depGroup) continue
+            if (indexer in alreadyExtracted) continue
+
+            // Collect full chain: walk up through dependsOn
+            val chain = mutableSetOf<Indexer>()
+            var current: Indexer? = indexer
+            while (current != null && current in indexerSet) {
+                chain.add(current)
+                current = current.dependsOn
+            }
+            // Collect downstream dependents
+            val dependentMap = indexers.filter { it.dependsOn != null }.groupBy { it.dependsOn!! }
+            val queue = ArrayDeque(chain.toList())
+            while (queue.isNotEmpty()) {
+                val node = queue.removeFirst()
+                dependentMap[node]?.forEach { dependent ->
+                    if (chain.add(dependent)) queue.addLast(dependent)
+                }
+            }
+
+            // Remove from original groups
+            for (member in chain) {
+                indexerToGroup[member]?.remove(member)
+                alreadyExtracted.add(member)
+            }
+            extracted.add(chain)
+        }
+
+        // Phase C: Re-merge extracted chains into closest group
+        for (chain in extracted) {
+            val chainMin = chain.minOf { it.getCurrentBlockNumber() }
+            var closestGroup: MutableSet<Indexer>? = null
+            var closestGap = Long.MAX_VALUE
+            for (group in groups) {
+                if (group.isEmpty()) continue
+                val groupMin = group.minOf { it.getCurrentBlockNumber() }
+                val gap = kotlin.math.abs(chainMin - groupMin)
+                if (gap <= threshold && gap < closestGap) {
+                    closestGap = gap
+                    closestGroup = group
+                }
+            }
+            if (closestGroup != null) {
+                closestGroup.addAll(chain)
+            } else {
+                groups.add(chain)
+            }
+        }
+
+        // Filter empty groups and topologically order each
+        return groups.filter { it.isNotEmpty() }.map { topologicalOrder(it.toList()).flatten() }
+    }
+
     private enum class VisitState {
         VISITING,
         VISITED,