feat: implement project locks containing multiple job ids

backend/api/src/main/kotlin/io/tolgee/api/v2/controllers/administration/ProjectBatchLockController.kt

@@ -47,14 +47,13 @@ class ProjectBatchLockController(
     logger.debug("Retrieving all project batch locks")
 
     val locks = batchJobProjectLockingManager.getMap()
-    val lockModels = locks.map { (projectId, lockedJobId) ->
-      val lockStatus = when (lockedJobId) {
-        null -> LockStatus.UNINITIALIZED
-        0L -> LockStatus.UNLOCKED
+    val lockModels = locks.entries.map { (projectId, lockedJobIds) ->
+      val lockStatus = when {
+        lockedJobIds.isEmpty() -> LockStatus.UNLOCKED
         else -> LockStatus.LOCKED
       }
 
-      val jobInfo = if (lockedJobId != null && lockedJobId > 0L) {
+      val jobInfos = lockedJobIds.mapNotNull { lockedJobId ->
         val jobDto = batchJobService.findJobDto(lockedJobId)
         if (jobDto == null) {
           logger.warn("Locked job $lockedJobId in project $projectId not found")
@@ -67,15 +66,13 @@ class ProjectBatchLockController(
             createdAt = jobDto.createdAt
           )
         }
-      } else {
-        null
       }
 
       ProjectLockModel(
         projectId = projectId,
-        lockedJobId = lockedJobId,
+        lockedJobIds = lockedJobIds,
         lockStatus = lockStatus,
-        jobInfo = jobInfo
+        jobInfos = jobInfos
       )
     }
 
@@ -113,9 +110,9 @@ class ProjectBatchLockController(
  */
 data class ProjectLockModel(
   val projectId: Long,
-  val lockedJobId: Long?,
+  val lockedJobIds: Set<Long>,
   val lockStatus: LockStatus,
-  val jobInfo: JobInfo?
+  val jobInfos: List<JobInfo>
 )
 
 /**
@@ -132,13 +129,10 @@ data class JobInfo(
  * Status of the project lock
  */
 enum class LockStatus {
-  /** Lock is explicitly cleared (value = 0L) */
+  /** Lock is explicitly cleared (value = empty set) */
   UNLOCKED,
 
-  /** Lock has never been initialized (value = null) */
-  UNINITIALIZED,
-
-  /** Lock is held by a specific job (value = jobId) */
+  /** Lock is held by one or more jobs (value = set of job IDs) */
   LOCKED
 }
 

backend/app/src/test/kotlin/io/tolgee/batch/BatchJobTestUtil.kt

@@ -409,7 +409,7 @@ class BatchJobTestUtil(
 
   fun verifyProjectJobLockReleased() {
     waitFor(pollTime = 200, timeout = 1000) {
-      batchJobProjectLockingManager.getLockedForProject(testData.projectBuilder.self.id) == 0L
+      batchJobProjectLockingManager.getLockedForProject(testData.projectBuilder.self.id).isEmpty()
     }
   }
 

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

@@ -170,7 +170,7 @@ class BatchJobConcurrentLauncher(
     }
 
     /**
-     * Only single job can run in project at the same time
+     * There is a project level lock with configurable n concurrent locks allowed.
      */
     if (!batchJobProjectLockingManager.canLockJobForProject(executionItem.jobId)) {
       logger.debug(

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

@@ -2,15 +2,18 @@ package io.tolgee.batch
 
 import io.tolgee.batch.data.BatchJobDto
 import io.tolgee.component.UsingRedisProvider
+import io.tolgee.configuration.tolgee.BatchProperties
 import io.tolgee.util.Logging
 import io.tolgee.util.logger
-import org.redisson.api.RMap
 import org.redisson.api.RedissonClient
+import org.springframework.beans.factory.InitializingBean
 import org.springframework.context.annotation.Lazy
 import org.springframework.stereotype.Component
 import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.ConcurrentMap
 
+private const val REDIS_PROJECT_BATCH_JOB_LOCKS_KEY = "project_batch_job_locks"
+
 /**
  * Only single job can be executed at the same time for one project.
  *
@@ -19,13 +22,14 @@ import java.util.concurrent.ConcurrentMap
 @Component
 class BatchJobProjectLockingManager(
   private val batchJobService: BatchJobService,
+  private val batchProperties: BatchProperties,
   @Lazy
   private val redissonClient: RedissonClient,
   private val usingRedisProvider: UsingRedisProvider,
-) : Logging {
+) : Logging, InitializingBean {
   companion object {
     private val localProjectLocks by lazy {
-      ConcurrentHashMap<Long, Long?>()
+      ConcurrentHashMap<Long, Set<Long>>()
     }
   }
 
@@ -51,18 +55,20 @@ class BatchJobProjectLockingManager(
     jobId: Long,
   ) {
     projectId ?: return
-    getMap().compute(projectId) { _, lockedJobId ->
+    getMap().compute(projectId) { _, lockedJobIds ->
       logger.debug("Unlocking job: $jobId for project $projectId")
-      if (lockedJobId == jobId) {
+      val currentJobs = lockedJobIds ?: emptySet()
+      if (currentJobs.contains(jobId)) {
         logger.debug("Unlocking job: $jobId for project $projectId")
-        return@compute 0L
+        val updatedJobs = currentJobs - jobId
+        return@compute updatedJobs.ifEmpty { emptySet() }
       }
       logger.debug("Job: $jobId for project $projectId is not locked")
-      return@compute lockedJobId
+      return@compute currentJobs
     }
   }
 
-  fun getMap(): ConcurrentMap<Long, Long?> {
+  fun getMap(): ConcurrentMap<Long, Set<Long>> {
     if (usingRedisProvider.areWeUsingRedis) {
       return getRedissonProjectLocks()
     }
@@ -71,65 +77,84 @@ class BatchJobProjectLockingManager(
 
   private fun tryLockWithRedisson(batchJobDto: BatchJobDto): Boolean {
     val projectId = batchJobDto.projectId ?: return true
-    val computed =
-      getRedissonProjectLocks().compute(projectId) { _, value ->
-        computeFnBody(batchJobDto, value)
-      }
-    return computed == batchJobDto.id
+    val computedJobIds =
+      getRedissonProjectLocks().compute(projectId) { _, lockedJobIds ->
+        val newLockedJobIds = computeFnBody(batchJobDto, lockedJobIds ?: emptySet())
+        logger.debug(
+          "While trying to lock on redis {} for project {} new lock value is {}",
+          batchJobDto.id,
+          batchJobDto.projectId,
+          newLockedJobIds
+        )
+        newLockedJobIds
+      } ?: emptySet()
+    return computedJobIds.contains(batchJobDto.id)
   }
 
-  fun getLockedForProject(projectId: Long): Long? {
+  fun getLockedForProject(projectId: Long): Set<Long> {
     if (usingRedisProvider.areWeUsingRedis) {
-      return getRedissonProjectLocks()[projectId]
+      return getRedissonProjectLocks()[projectId] ?: emptySet()
     }
-    return localProjectLocks[projectId]
+    return localProjectLocks[projectId] ?: emptySet()
   }
 
-  private fun tryLockLocal(toLock: BatchJobDto): Boolean {
-    val projectId = toLock.projectId ?: return true
-    val computed =
-      localProjectLocks.compute(projectId) { _, value ->
-        val newLocked = computeFnBody(toLock, value)
-        logger.debug("While trying to lock ${toLock.id} for project ${toLock.projectId} new lock value is $newLocked")
-        newLocked
-      }
-    return computed == toLock.id
+  private fun tryLockLocal(batchJobDto: BatchJobDto): Boolean {
+    val projectId = batchJobDto.projectId ?: return true
+    val computedJobIds =
+      localProjectLocks.compute(projectId) { _, lockedJobIds ->
+        val newLockedJobIds = computeFnBody(batchJobDto, lockedJobIds ?: emptySet())
+        logger.debug(
+          "While trying to lock locally {} for project {} new lock value is {}",
+          batchJobDto.id,
+          batchJobDto.projectId,
+          newLockedJobIds
+        )
+        newLockedJobIds
+      } ?: emptySet()
+    return computedJobIds.contains(batchJobDto.id)
   }
 
   private fun computeFnBody(
     toLock: BatchJobDto,
-    currentValue: Long?,
-  ): Long {
+    lockedJobIds: Set<Long>,
+  ): Set<Long> {
     val projectId =
       toLock.projectId
         ?: throw IllegalStateException(
           "Project id is required. " +
             "Locking for project should not happen for non-project jobs.",
         )
-    // nothing is locked
-    if (currentValue == 0L) {
-      logger.debug("Locking job ${toLock.id} for project ${toLock.projectId}, nothing is locked")
-      return toLock.id
-    }
 
-    // value for the project is not initialized yet
-    if (currentValue == null) {
+    // nothing is locked
+    if (lockedJobIds.isEmpty()) {
       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)
-      logger.debug("Initial locked job $initial for project ${toLock.projectId}")
-      if (initial == null) {
-        logger.debug("No job found, locking ${toLock.id}")
-        return toLock.id
+      val initialJobId = getInitialJobId(projectId)
+      logger.info("Initial locked job $initialJobId for project ${toLock.projectId}")
+      if (initialJobId == null) {
+        logger.debug("No initial job found, locking only ${toLock.id}")
+        return setOf(toLock.id)
       }
+      val newLockedJobIds = mutableSetOf<Long>(initialJobId)
+      if (newLockedJobIds.size < batchProperties.projectConcurrency) {
+        logger.debug("Locking job ${toLock.id} for project $projectId. Active jobs before: $newLockedJobIds")
+        newLockedJobIds.add(toLock.id)
+      } else {
+        logger.debug(
+          "Cannot lock job ${toLock.id} for project $projectId, limit reached. Active jobs: $newLockedJobIds"
+        )
+      }
+      return newLockedJobIds
+    }
 
-      logger.debug("Job found, locking $initial")
-      return initial
+    // standard case - there are some jobs locked
+    if (lockedJobIds.size < batchProperties.projectConcurrency) {
+      logger.debug("Locking job ${toLock.id} for project $projectId. Active jobs before: $lockedJobIds")
+      return lockedJobIds + toLock.id
     }
 
-    logger.debug("Job $currentValue is locked for project ${toLock.projectId}")
     // if we cannot lock, we are returning current value
-    return currentValue
+    return lockedJobIds
   }
 
   private fun getInitialJobId(projectId: Long): Long? {
@@ -161,11 +186,47 @@ class BatchJobProjectLockingManager(
     return null
   }
 
-  private fun getRedissonProjectLocks(): RMap<Long, Long> {
-    return redissonClient.getMap("project_batch_job_locks")
+  private fun getRedissonProjectLocks(): ConcurrentMap<Long, Set<Long>> {
+      return redissonClient.getMap(REDIS_PROJECT_BATCH_JOB_LOCKS_KEY)
+    }
+
+  override fun afterPropertiesSet() {
+    // This runs first to check if redis has a map of the old format.
+    // If so, we migrate it to the new format.
+    if (!usingRedisProvider.areWeUsingRedis) {
+      logger.debug("Not using Redis, skipping migration check")
+      return
+    }
+
+    val redisProjectBatchJobLocks = redissonClient.getMap<Long, Any>(REDIS_PROJECT_BATCH_JOB_LOCKS_KEY)
+    val isOldFormat = redisProjectBatchJobLocks.values.any { it is Long || it == null }
+    if (!isOldFormat) {
+      logger.debug("Redis project locks are in new format, no migration needed")
+      return
+    }
+
+    logger.info("Starting migration of project locks from old format (v1) to new format (v2)")
+    // First, copy all data from Redis to local memory
+    val localCopy = mutableMapOf<Long, Set<Long>>()
+    redisProjectBatchJobLocks.forEach { (projectId, jobId) ->
+      val jobSet = when (jobId) {
+        null, 0L -> emptySet<Long>()
+        else -> setOf<Long>(jobId as Long)
+      }
+      localCopy[projectId] = jobSet
+    }
+    logger.info("Copied ${localCopy.size} project locks from old format to local memory")
+
+    // Write all data back in new format (this will overwrite the old format)
+    val newMap = getRedissonProjectLocks()
+    localCopy.forEach { (projectId, jobSet) ->
+      newMap[projectId] = jobSet
+    }
+
+    logger.info("Successfully migrated ${newMap.size} project locks from local memory to new format")
   }
 
   fun getLockedJobIds(): Set<Long> {
-    return getMap().values.filterNotNull().toSet()
+    return getMap().values.flatten().toSet()
   }
 }

backend/data/src/main/kotlin/io/tolgee/configuration/tolgee/BatchProperties.kt

@@ -11,4 +11,7 @@ class BatchProperties {
 
   @DocProperty(description = "How many job chunks are added to the internal queue on each scheduled run")
   var chunkQueuePopulationSize: Int = 1_000
+
+  @DocProperty(description = "How many parallel jobs can be run at once per project across all Tolgee instances")
+  var projectConcurrency: Int = 1
 }