feat: optimize batch job throughput

backend/data/src/main/kotlin/io/tolgee/batch/BatchJobChunkExecutionQueue.kt

@@ -5,6 +5,7 @@ import io.tolgee.Metrics
 import io.tolgee.batch.data.BatchJobChunkExecutionDto
 import io.tolgee.batch.data.ExecutionQueueItem
 import io.tolgee.batch.data.QueueEventType
+import io.tolgee.batch.timing.BatchJobTimerProvider
 import io.tolgee.batch.events.JobQueueItemsEvent
 import io.tolgee.component.UsingRedisProvider
 import io.tolgee.configuration.tolgee.BatchProperties
@@ -36,6 +37,7 @@ class BatchJobChunkExecutionQueue(
   @Lazy
   private val redisTemplate: StringRedisTemplate,
   private val metrics: Metrics,
+  private val timer: BatchJobTimerProvider?,
 ) : Logging,
   InitializingBean {
   companion object {
@@ -57,6 +59,11 @@ class BatchJobChunkExecutionQueue(
     private var lastServedJobId: Long? = null
   }
 
+  private fun <T> timed(
+    name: String,
+    block: () -> T,
+  ): T = timer?.measure(name, block) ?: block()
+
   private fun incrementCharacterCount(character: JobCharacter) {
     jobCharacterCounts.computeIfAbsent(character) { AtomicInteger(0) }.incrementAndGet()
   }
@@ -75,9 +82,11 @@ class BatchJobChunkExecutionQueue(
       QueueEventType.REMOVE -> {
         // Remove and decrement atomically per item to prevent double-decrement
         // if poll() removes an item between removeAll and forEach
-        event.items.forEach { item ->
-          if (queue.remove(item)) {
-            decrementCharacterCount(item.jobCharacter)
+        timed("QUEUE_INTERNAL.removeConsuming") {
+          event.items.forEach { item ->
+            if (queue.remove(item)) {
+              decrementCharacterCount(item.jobCharacter)
+            }
           }
         }
       }
@@ -117,7 +126,10 @@ class BatchJobChunkExecutionQueue(
   }
 
   fun addExecutionsToLocalQueue(data: List<BatchJobChunkExecutionDto>) {
-    val ids = queue.map { it.chunkExecutionId }.toSet()
+    val ids =
+      timed("QUEUE_INTERNAL.buildIdSet") {
+        queue.map { it.chunkExecutionId }.toSet()
+      }
     var count = 0
     data.forEach {
       if (!ids.contains(it.id)) {
@@ -133,7 +145,10 @@ class BatchJobChunkExecutionQueue(
 
   fun addItemsToLocalQueue(data: List<ExecutionQueueItem>) {
     // Use Set for O(1) lookup instead of O(n) queue.contains()
-    val existingIds = queue.mapTo(HashSet()) { it.chunkExecutionId }
+    val existingIds =
+      timed("QUEUE_INTERNAL.buildIdSet") {
+        queue.mapTo(HashSet()) { it.chunkExecutionId }
+      }
     val toAdd = mutableListOf<ExecutionQueueItem>()
     var filteredOutCount = 0
 
@@ -188,12 +203,14 @@ class BatchJobChunkExecutionQueue(
 
   fun removeJobExecutions(jobId: Long) {
     logger.debug("Removing job $jobId from queue, queue size: ${queue.size}")
-    val iterator = queue.iterator()
-    while (iterator.hasNext()) {
-      val item = iterator.next()
-      if (item.jobId == jobId) {
-        iterator.remove()
-        decrementCharacterCount(item.jobCharacter)
+    timed("QUEUE_INTERNAL.removeJobExecutions") {
+      val iterator = queue.iterator()
+      while (iterator.hasNext()) {
+        val item = iterator.next()
+        if (item.jobId == jobId) {
+          iterator.remove()
+          decrementCharacterCount(item.jobCharacter)
+        }
       }
     }
     logger.debug("Removed job $jobId from queue, queue size: ${queue.size}")
@@ -237,8 +254,11 @@ class BatchJobChunkExecutionQueue(
       return null
     }
 
-    // Get distinct job IDs in queue order
-    val jobIds = queue.mapTo(LinkedHashSet()) { it.jobId }.toList()
+    // Get distinct job IDs in queue order — O(n) scan of entire queue
+    val jobIds =
+      timed("QUEUE_INTERNAL.collectJobIds") {
+        queue.mapTo(LinkedHashSet()) { it.jobId }.toList()
+      }
     if (jobIds.isEmpty()) {
       return null
     }
@@ -258,13 +278,22 @@ class BatchJobChunkExecutionQueue(
       val jobIndex = (startIndex + i) % jobIds.size
       val targetJobId = jobIds[jobIndex]
 
-      // Find first item for this job and try to remove it
-      val item = queue.firstOrNull { it.jobId == targetJobId }
-      if (item != null && queue.remove(item)) {
-        // Successfully removed - update state and return
-        decrementCharacterCount(item.jobCharacter)
-        lastServedJobId = targetJobId
-        return item
+      // Find first item for this job — O(n) scan to find matching item
+      val item =
+        timed("QUEUE_INTERNAL.findFirstForJob") {
+          queue.firstOrNull { it.jobId == targetJobId }
+        }
+      if (item != null) {
+        // Remove item — O(n) scan for ConcurrentLinkedQueue.remove()
+        val removed =
+          timed("QUEUE_INTERNAL.removeItem") {
+            queue.remove(item)
+          }
+        if (removed) {
+          decrementCharacterCount(item.jobCharacter)
+          lastServedJobId = targetJobId
+          return item
+        }
       }
       // Item was null or already removed by another thread, try next job
     }
@@ -298,7 +327,9 @@ class BatchJobChunkExecutionQueue(
   }
 
   fun getQueuedJobItems(jobId: Long): List<ExecutionQueueItem> {
-    return queue.filter { it.jobId == jobId }
+    return timed("QUEUE_INTERNAL.getQueuedJobItems") {
+      queue.filter { it.jobId == jobId }
+    }
   }
 
   fun getAllQueueItems(): List<ExecutionQueueItem> {

backend/data/src/main/kotlin/io/tolgee/batch/BatchJobConcurrentLauncher.kt

@@ -3,6 +3,7 @@ package io.tolgee.batch
 import io.sentry.Sentry
 import io.tolgee.batch.data.BatchJobDto
 import io.tolgee.batch.data.ExecutionQueueItem
+import io.tolgee.batch.timing.BatchJobTimerProvider
 import io.tolgee.component.CurrentDateProvider
 import io.tolgee.configuration.tolgee.BatchProperties
 import io.tolgee.fixtures.waitFor
@@ -31,6 +32,7 @@ class BatchJobConcurrentLauncher(
   private val progressManager: ProgressManager,
   private val batchJobActionService: BatchJobActionService,
   private val tracingContext: TolgeeTracingContext,
+  private val timerProvider: BatchJobTimerProvider?,
 ) : Logging {
   companion object {
     const val MIN_TIME_BETWEEN_OPERATIONS = 100
@@ -203,7 +205,11 @@ class BatchJobConcurrentLauncher(
     /**
      * Only single job can run in project at the same time
      */
-    if (!batchJobProjectLockingManager.canLockJobForProject(executionItem.jobId)) {
+    val canLock =
+      timed("LAUNCHER.projectLockCheck") {
+        batchJobProjectLockingManager.canLockJobForProject(executionItem.jobId)
+      }
+    if (!canLock) {
       logger.debug(
         "⚠️ Cannot run execution ${executionItem.chunkExecutionId}. " +
           "Other job from the project is currently running, skipping",
@@ -299,6 +305,13 @@ class BatchJobConcurrentLauncher(
     return runningCount < allowedCharacterCounts
   }
 
+  private fun <T> timed(
+    operationName: String,
+    block: () -> T,
+  ): T {
+    return timerProvider?.measure(operationName, block) ?: block()
+  }
+
   private fun ExecutionQueueItem.trySetRunningState(batchJobDto: BatchJobDto): Boolean {
     // Check maxPerJobConcurrency before trying to set running state
     val maxPerJobConcurrency = batchJobDto.maxPerJobConcurrency

backend/data/src/main/kotlin/io/tolgee/batch/BatchJobProjectLockingManager.kt

@@ -1,6 +1,7 @@
 package io.tolgee.batch
 
 import io.tolgee.batch.data.BatchJobDto
+import io.tolgee.batch.timing.BatchJobTimerProvider
 import io.tolgee.component.UsingRedisProvider
 import io.tolgee.util.Logging
 import io.tolgee.util.logger
@@ -22,27 +23,43 @@ class BatchJobProjectLockingManager(
   @Lazy
   private val redissonClient: RedissonClient,
   private val usingRedisProvider: UsingRedisProvider,
+  private val timerProvider: BatchJobTimerProvider?,
 ) : Logging {
   companion object {
     private val localProjectLocks by lazy {
       ConcurrentHashMap<Long, Long?>()
     }
+
+    /**
+     * Local cache of Redis lock state per project, to avoid redundant Redis round-trips.
+     * Key: projectId, Value: (lockedJobId, timestampMs).
+     * Entries expire after [LOCK_CACHE_TTL_MS] and are eagerly invalidated on unlock.
+     */
+    private val redisLockCache = ConcurrentHashMap<Long, LockCacheEntry>()
+    private const val LOCK_CACHE_TTL_MS = 1000L
   }
 
+  private data class LockCacheEntry(
+    val lockedJobId: Long,
+    val timestamp: Long,
+  )
+
   fun canLockJobForProject(batchJobId: Long): Boolean {
-    val jobDto = batchJobService.getJobDto(batchJobId)
-    if (!jobDto.type.exclusive) {
-      return true
+    return timed("PROJECT_LOCK.canLock") {
+      val jobDto = batchJobService.getJobDto(batchJobId)
+      if (!jobDto.type.exclusive) {
+        return@timed true
+      }
+      tryLockJobForProject(jobDto)
     }
-    return tryLockJobForProject(jobDto)
   }
 
   private fun tryLockJobForProject(jobDto: BatchJobDto): Boolean {
     logger.debug("Trying to lock job ${jobDto.id} for project ${jobDto.projectId}")
     return if (usingRedisProvider.areWeUsingRedis) {
-      tryLockWithRedisson(jobDto)
+      timed("PROJECT_LOCK.tryLockRedis") { tryLockWithRedisson(jobDto) }
     } else {
-      tryLockLocal(jobDto)
+      timed("PROJECT_LOCK.tryLockLocal") { tryLockLocal(jobDto) }
     }
   }
 
@@ -51,14 +68,18 @@ class BatchJobProjectLockingManager(
     jobId: Long,
   ) {
     projectId ?: return
-    getMap().compute(projectId) { _, lockedJobId ->
-      logger.debug("Unlocking job: $jobId for project $projectId")
-      if (lockedJobId == jobId) {
+    timed("PROJECT_LOCK.unlock") {
+      // Eagerly invalidate the local cache so other jobs can acquire the lock immediately
+      redisLockCache.remove(projectId)
+      getMap().compute(projectId) { _, lockedJobId ->
         logger.debug("Unlocking job: $jobId for project $projectId")
-        return@compute 0L
+        if (lockedJobId == jobId) {
+          logger.debug("Unlocking job: $jobId for project $projectId")
+          return@compute 0L
+        }
+        logger.debug("Job: $jobId for project $projectId is not locked")
+        return@compute lockedJobId
       }
-      logger.debug("Job: $jobId for project $projectId is not locked")
-      return@compute lockedJobId
     }
   }
 
@@ -71,10 +92,24 @@ class BatchJobProjectLockingManager(
 
   private fun tryLockWithRedisson(batchJobDto: BatchJobDto): Boolean {
     val projectId = batchJobDto.projectId ?: return true
+
+    // Check local cache first to avoid Redis round-trip for the common "still locked" case
+    val cached = redisLockCache[projectId]
+    if (cached != null &&
+      cached.lockedJobId != batchJobDto.id &&
+      cached.lockedJobId != 0L &&
+      System.currentTimeMillis() - cached.timestamp < LOCK_CACHE_TTL_MS
+    ) {
+      timed("PROJECT_LOCK.cacheHit") {}
+      return false
+    }
+
     val computed =
       getRedissonProjectLocks().compute(projectId) { _, value ->
         computeFnBody(batchJobDto, value)
       }
+    // Update local cache with the result
+    redisLockCache[projectId] = LockCacheEntry(computed ?: 0L, System.currentTimeMillis())
     return computed == batchJobDto.id
   }
 
@@ -116,7 +151,7 @@ class BatchJobProjectLockingManager(
     if (currentValue == null) {
       logger.debug("Getting initial locked state from DB state")
       // we have to find out from database if there is any running job for the project
-      val initial = getInitialJobId(projectId)
+      val initial = timed("PROJECT_LOCK.getInitialJobId") { getInitialJobId(projectId) }
       logger.debug("Initial locked job $initial for project ${toLock.projectId}")
       if (initial == null) {
         logger.debug("No job found, locking ${toLock.id}")
@@ -171,4 +206,11 @@ class BatchJobProjectLockingManager(
   fun getLockedJobIds(): Set<Long> {
     return getMap().values.filterNotNull().toSet()
   }
+
+  private fun <T> timed(
+    operationName: String,
+    block: () -> T,
+  ): T {
+    return timerProvider?.measure(operationName, block) ?: block()
+  }
 }

backend/data/src/main/kotlin/io/tolgee/batch/data/BatchJobType.kt

@@ -114,4 +114,9 @@ enum class BatchJobType(
     processor = NoOpChunkProcessor::class,
     exclusive = false,
   ),
+  NO_OP_EXCLUSIVE(
+    activityType = null,
+    maxRetries = 0,
+    processor = NoOpChunkProcessor::class,
+  ),
 }

backend/data/src/main/kotlin/io/tolgee/batch/processors/NoOpChunkProcessor.kt

@@ -7,25 +7,32 @@ import io.tolgee.batch.request.NoOpRequest
 import org.springframework.stereotype.Component
 import kotlin.coroutines.CoroutineContext
 
+data class NoOpParams(
+  val chunkProcessingDelayMs: Long = 0,
+)
+
 @Component
 class NoOpChunkProcessor(
   private val progressManager: ProgressManager,
-) : ChunkProcessor<NoOpRequest, Any?, Long> {
+) : ChunkProcessor<NoOpRequest, NoOpParams, Long> {
   override fun process(
     job: BatchJobDto,
     chunk: List<Long>,
     coroutineContext: CoroutineContext,
   ) {
-    // Report progress for the whole chunk at once
+    val params = getParams(job)
+    if (params.chunkProcessingDelayMs > 0) {
+      Thread.sleep(params.chunkProcessingDelayMs)
+    }
     progressManager.reportSingleChunkProgress(job.id, chunk.size)
   }
 
-  override fun getParamsType(): Class<Any?>? {
-    return null
+  override fun getParamsType(): Class<NoOpParams> {
+    return NoOpParams::class.java
   }
 
-  override fun getParams(data: NoOpRequest): Any? {
-    return null
+  override fun getParams(data: NoOpRequest): NoOpParams {
+    return NoOpParams(chunkProcessingDelayMs = data.chunkProcessingDelayMs)
   }
 
   override fun getTargetItemType(): Class<Long> {

backend/data/src/main/kotlin/io/tolgee/batch/request/NoOpRequest.kt

@@ -5,4 +5,12 @@ import jakarta.validation.constraints.NotEmpty
 class NoOpRequest {
   @NotEmpty
   var itemIds: List<Long> = listOf()
+  var chunkProcessingDelayMs: Long = 0
+}
+
+class NoOpMultiRequest {
+  var totalItems: Int = 10000
+  var numberOfJobs: Int = 1
+  var chunkProcessingDelayMs: Long = 0
+  var numberOfProjects: Int = 0
 }

backend/data/src/main/kotlin/io/tolgee/batch/timing/BatchJobTimerProvider.kt

@@ -0,0 +1,13 @@
+package io.tolgee.batch.timing
+
+/**
+ * Interface for inline timing of internal operations.
+ * Implemented by BatchJobOperationTimer in the development module.
+ * When not available (production), operations run without timing overhead.
+ */
+interface BatchJobTimerProvider {
+  fun <T> measure(
+    operationName: String,
+    block: () -> T,
+  ): T
+}

backend/development/build.gradle

@@ -66,6 +66,7 @@ dependencies {
     implementation 'org.springframework.boot:spring-boot-starter-web'
     implementation 'org.springframework.boot:spring-boot-starter-data-jpa'
     implementation 'org.springframework.boot:spring-boot-starter-hateoas'
+    implementation 'org.springframework.boot:spring-boot-starter-aop'
     implementation "org.springframework.boot:spring-boot-configuration-processor"
     implementation project(":data")
     implementation project(":api")