diff --git a/backend/api/src/main/kotlin/io/tolgee/websocket/RedisWebsocketEventPublisher.kt b/backend/api/src/main/kotlin/io/tolgee/websocket/RedisWebsocketEventPublisher.kt index 135aadea55..fe99453f90 100644 --- a/backend/api/src/main/kotlin/io/tolgee/websocket/RedisWebsocketEventPublisher.kt +++ b/backend/api/src/main/kotlin/io/tolgee/websocket/RedisWebsocketEventPublisher.kt @@ -1,16 +1,17 @@ package io.tolgee.websocket -import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper +import com.fasterxml.jackson.databind.ObjectMapper import org.springframework.data.redis.core.StringRedisTemplate class RedisWebsocketEventPublisher( private val redisTemplate: StringRedisTemplate, + private val objectMapper: ObjectMapper, ) : WebsocketEventPublisher { override operator fun invoke( destination: String, message: WebsocketEvent, ) { - val messageString = jacksonObjectMapper().writeValueAsString(RedisWebsocketEventWrapper(destination, message)) + val messageString = objectMapper.writeValueAsString(RedisWebsocketEventWrapper(destination, message)) redisTemplate.convertAndSend( "websocket", messageString, diff --git a/backend/api/src/main/kotlin/io/tolgee/websocket/WebsocketPublisherConfiguration.kt b/backend/api/src/main/kotlin/io/tolgee/websocket/WebsocketPublisherConfiguration.kt index e5d1b3b1e1..9bf08b4a1c 100644 --- a/backend/api/src/main/kotlin/io/tolgee/websocket/WebsocketPublisherConfiguration.kt +++ b/backend/api/src/main/kotlin/io/tolgee/websocket/WebsocketPublisherConfiguration.kt @@ -1,5 +1,6 @@ package io.tolgee.websocket +import com.fasterxml.jackson.databind.ObjectMapper import io.tolgee.configuration.tolgee.WebsocketProperties import org.springframework.context.ApplicationContext import org.springframework.context.annotation.Bean @@ -15,7 +16,10 @@ class WebsocketPublisherConfiguration( @Bean fun websocketEventPublisher(): WebsocketEventPublisher { if (websocketProperties.useRedis) { - return RedisWebsocketEventPublisher(applicationContext.getBean(StringRedisTemplate::class.java)) + return RedisWebsocketEventPublisher( + applicationContext.getBean(StringRedisTemplate::class.java), + applicationContext.getBean(ObjectMapper::class.java), + ) } return SimpleWebsocketEventPublisher(applicationContext.getBean(SimpMessagingTemplate::class.java)) } diff --git a/backend/app/src/test/kotlin/io/tolgee/batch/state/RedisBatchJobStateStorageTest.kt b/backend/app/src/test/kotlin/io/tolgee/batch/state/RedisBatchJobStateStorageTest.kt new file mode 100644 index 0000000000..3af91ec08d --- /dev/null +++ b/backend/app/src/test/kotlin/io/tolgee/batch/state/RedisBatchJobStateStorageTest.kt @@ -0,0 +1,165 @@ +package io.tolgee.batch.state + +import io.tolgee.AbstractSpringTest +import io.tolgee.fixtures.RedisRunner +import io.tolgee.model.batch.BatchJobChunkExecutionStatus +import io.tolgee.testing.ContextRecreatingTest +import io.tolgee.testing.assert +import org.junit.jupiter.api.AfterAll +import org.junit.jupiter.api.AfterEach +import org.junit.jupiter.api.Test +import org.mockito.kotlin.any +import org.mockito.kotlin.clearInvocations +import org.mockito.kotlin.never +import org.mockito.kotlin.times +import org.mockito.kotlin.verify +import org.redisson.api.RedissonClient +import org.redisson.client.codec.StringCodec +import org.springframework.beans.factory.annotation.Autowired +import org.springframework.boot.test.context.SpringBootTest +import org.springframework.boot.test.util.TestPropertyValues +import org.springframework.context.ApplicationContextInitializer +import org.springframework.context.ConfigurableApplicationContext +import org.springframework.test.annotation.DirtiesContext +import org.springframework.test.context.ContextConfiguration +import org.springframework.test.context.bean.override.mockito.MockitoSpyBean + +@ContextRecreatingTest +@SpringBootTest( + properties = [ + "tolgee.cache.use-redis=true", + "tolgee.cache.enabled=true", + "tolgee.websocket.use-redis=true", + ], +) +@ContextConfiguration(initializers = [RedisBatchJobStateStorageTest.Companion.Initializer::class]) +@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_CLASS) +class RedisBatchJobStateStorageTest : AbstractSpringTest() { + companion object { + private const val STATE_KEY_PREFIX = "batch_job_state:" + private const val STATE_INITIALIZED_PREFIX = "batch_job_state_initialized:" + private const val STARTED_PREFIX = "batch_job_started:" + + val redisRunner = RedisRunner() + + @AfterAll + @JvmStatic + fun stopRedis() { + redisRunner.stop() + } + + class Initializer : ApplicationContextInitializer { + override fun initialize(configurableApplicationContext: ConfigurableApplicationContext) { + redisRunner.run() + TestPropertyValues + .of("spring.data.redis.port=${RedisRunner.port}") + .applyTo(configurableApplicationContext) + } + } + } + + @Autowired + lateinit var stateProvider: BatchJobStateProvider + + @MockitoSpyBean + @Autowired + lateinit var redissonClient: RedissonClient + + private val usedJobIds = mutableSetOf() + + @AfterEach + fun cleanup() { + usedJobIds.forEach { stateProvider.removeJobState(it) } + usedJobIds.clear() + } + + @Test + fun `removeJobState pipelines its deletes into a single RBatch`() { + val jobId = + seed(900_020L, BatchJobChunkExecutionStatus.SUCCESS) { + incrementRunningCount(it) + tryMarkJobStarted(it) + } + + clearInvocations(redissonClient) + stateProvider.removeJobState(jobId) + + // The deletes must be pipelined through one RBatch, not issued as per-key round-trips. + verify(redissonClient, times(1)).createBatch() + verify(redissonClient, never()).getAtomicLong(any()) + + stateProvider.hasCachedJobState(jobId).assert.isFalse() + } + + @Test + fun `clearUnusedStates clears completed jobs across chunks and keeps counters and unfinished jobs`() { + // Exceeds the production CLEANUP_BATCH_SIZE (100) so cleanup runs across more than one chunk. + val completedJobIds = (901_000L until 901_150L).toList() + completedJobIds.forEach { + seed(it, BatchJobChunkExecutionStatus.SUCCESS) { + incrementRunningCount(it) + tryMarkJobStarted(it) + } + } + val runningJobId = seed(900_002L, BatchJobChunkExecutionStatus.RUNNING) + + stateProvider.clearUnusedStates() + + // Every completed job's hash is cleared — including those in the second chunk. + completedJobIds.forEach { stateProvider.hasCachedJobState(it).assert.isFalse() } + + // Counters and the started marker survive until removeJobState finalizes the job + // (dropping the started marker would allow re-execution). + val sample = completedJobIds.first() + isInitializedBucketPresent(sample).assert.isFalse() + stateProvider.getRunningCount(sample).assert.isEqualTo(1) + isStartedBucketPresent(sample).assert.isTrue() + + // Unfinished job is left untouched. + stateProvider.hasCachedJobState(runningJobId).assert.isTrue() + isInitializedBucketPresent(runningJobId).assert.isTrue() + } + + @Test + fun `clearUnusedStates skips a job whose state cannot be read and still cleans healthy jobs`() { + val healthyJobId = seed(900_040L, BatchJobChunkExecutionStatus.SUCCESS) + val corruptJobId = 900_041L + usedJobIds += corruptJobId + // Forge a value the state codec cannot deserialize, so this job's read future fails. + redissonClient.getMap("$STATE_KEY_PREFIX$corruptJobId", StringCodec.INSTANCE)["1"] = "corrupt" + redissonClient.getBucket("$STATE_INITIALIZED_PREFIX$corruptJobId").set(true) + + stateProvider.clearUnusedStates() + + // The unreadable job is skipped, not deleted... + stateProvider.hasCachedJobState(corruptJobId).assert.isTrue() + // ...while the healthy completed job in the same pass is still cleaned. + stateProvider.hasCachedJobState(healthyJobId).assert.isFalse() + } + + private fun seed( + jobId: Long, + status: BatchJobChunkExecutionStatus, + counters: BatchJobStateProvider.(Long) -> Unit = {}, + ): Long { + usedJobIds += jobId + stateProvider.updateSingleExecution(jobId, 1L, executionState(status)) + redissonClient.getBucket("$STATE_INITIALIZED_PREFIX$jobId").set(true) + stateProvider.counters(jobId) + return jobId + } + + private fun executionState(status: BatchJobChunkExecutionStatus) = + ExecutionState( + successTargetsCount = 0, + status = status, + chunkNumber = 0, + retry = null, + transactionCommitted = true, + ) + + private fun isInitializedBucketPresent(jobId: Long) = + redissonClient.getBucket("$STATE_INITIALIZED_PREFIX$jobId").isExists + + private fun isStartedBucketPresent(jobId: Long) = redissonClient.getBucket("$STARTED_PREFIX$jobId").isExists +} diff --git a/backend/data/src/main/kotlin/io/tolgee/batch/BatchJobCancellationManager.kt b/backend/data/src/main/kotlin/io/tolgee/batch/BatchJobCancellationManager.kt index f083b31ac3..be601c6239 100644 --- a/backend/data/src/main/kotlin/io/tolgee/batch/BatchJobCancellationManager.kt +++ b/backend/data/src/main/kotlin/io/tolgee/batch/BatchJobCancellationManager.kt @@ -1,6 +1,6 @@ package io.tolgee.batch -import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper +import com.fasterxml.jackson.databind.ObjectMapper import io.tolgee.activity.ActivityHolder import io.tolgee.batch.cleaning.BatchJobStatusProvider import io.tolgee.batch.events.JobCancelEvent @@ -36,6 +36,7 @@ class BatchJobCancellationManager( private val batchJobChunkExecutionQueue: BatchJobChunkExecutionQueue, private val concurrentExecutionLauncher: BatchJobConcurrentLauncher, private val batchProperties: io.tolgee.configuration.tolgee.BatchProperties, + private val objectMapper: ObjectMapper, ) : Logging { @Transactional fun cancel(id: Long) { @@ -63,7 +64,7 @@ class BatchJobCancellationManager( if (usingRedisProvider.areWeUsingRedis) { redisTemplate.convertAndSend( RedisPubSubReceiverConfiguration.JOB_CANCEL_TOPIC, - jacksonObjectMapper().writeValueAsString(id), + objectMapper.writeValueAsString(id), ) } cancelLocalJob(id) diff --git a/backend/data/src/main/kotlin/io/tolgee/batch/BatchJobChunkExecutionQueue.kt b/backend/data/src/main/kotlin/io/tolgee/batch/BatchJobChunkExecutionQueue.kt index 020dd2bed6..12824be23e 100644 --- a/backend/data/src/main/kotlin/io/tolgee/batch/BatchJobChunkExecutionQueue.kt +++ b/backend/data/src/main/kotlin/io/tolgee/batch/BatchJobChunkExecutionQueue.kt @@ -1,6 +1,6 @@ package io.tolgee.batch -import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper +import com.fasterxml.jackson.databind.ObjectMapper import io.tolgee.Metrics import io.tolgee.batch.data.BatchJobChunkExecutionDto import io.tolgee.batch.data.BatchJobType @@ -35,6 +35,7 @@ class BatchJobChunkExecutionQueue( @Lazy private val redisTemplate: StringRedisTemplate, private val metrics: Metrics, + private val objectMapper: ObjectMapper, ) : Logging, InitializingBean { companion object { @@ -222,7 +223,7 @@ class BatchJobChunkExecutionQueue( val event = JobQueueItemsEvent(batch, QueueEventType.ADD) redisTemplate.convertAndSend( RedisPubSubReceiverConfiguration.JOB_QUEUE_TOPIC, - jacksonObjectMapper().writeValueAsString(event), + objectMapper.writeValueAsString(event), ) } return diff --git a/backend/data/src/main/kotlin/io/tolgee/batch/state/RedisBatchJobStateStorage.kt b/backend/data/src/main/kotlin/io/tolgee/batch/state/RedisBatchJobStateStorage.kt index 01adee5667..1663e478c5 100644 --- a/backend/data/src/main/kotlin/io/tolgee/batch/state/RedisBatchJobStateStorage.kt +++ b/backend/data/src/main/kotlin/io/tolgee/batch/state/RedisBatchJobStateStorage.kt @@ -5,6 +5,7 @@ import io.tolgee.model.batch.BatchJobChunkExecution import io.tolgee.util.Logging import io.tolgee.util.logger import org.redisson.api.RAtomicLong +import org.redisson.api.RBatch import org.redisson.api.RMap import org.redisson.api.RedissonClient import java.util.concurrent.ConcurrentHashMap @@ -45,6 +46,8 @@ open class RedisBatchJobStateStorage( private const val REDIS_CANCELLED_COUNT_KEY_PREFIX = "batch_job_cancelled:" private const val REDIS_COMMITTED_COUNT_KEY_PREFIX = "batch_job_committed:" private const val REDIS_STARTED_KEY_PREFIX = "batch_job_started:" + + private const val CLEANUP_BATCH_SIZE = 100 } // Local cache for initialization status - avoids Redis calls for already-initialized jobs @@ -175,7 +178,7 @@ open class RedisBatchJobStateStorage( } override fun tryMarkJobStarted(jobId: Long): Boolean { - val bucket = redissonClient.getBucket("$REDIS_STARTED_KEY_PREFIX$jobId") + val bucket = redissonClient.getBucket(startedKey(jobId)) return bucket.setIfAbsent(true) } @@ -192,13 +195,12 @@ open class RedisBatchJobStateStorage( override fun removeJobState(jobId: Long) { logger.debug("Removing job state for job $jobId") - removeAllCounters(jobId) - val redisHash = getRedisHashForJob(jobId) - redisHash.delete() - // Also remove initialization and started markers - redissonClient.getBucket("$REDIS_STATE_INITIALIZED_KEY_PREFIX$jobId").delete() - redissonClient.getBucket("$REDIS_STARTED_KEY_PREFIX$jobId").delete() - // Clear local initialization cache to allow re-initialization if jobId is reused + val batch = redissonClient.createBatch() + addCounterDeletesToBatch(batch, jobId) + batch.getMap(stateKey(jobId)).deleteAsync() + batch.getBucket(initializedKey(jobId)).deleteAsync() + batch.getBucket(startedKey(jobId)).deleteAsync() + batch.execute() localInitializedJobs.remove(jobId) } @@ -213,25 +215,52 @@ open class RedisBatchJobStateStorage( /** * Cleans up batch job state hashes where all executions have a completed status. - * - * This uses HVALS to read all values from each hash. Because [ExecutionState] only - * stores lightweight metadata (no successTargets list), each value is ~50 bytes, - * making this operation fast even for jobs with thousands of chunks. */ override fun clearUnusedStates() { - val keys = redissonClient.keys.getKeysByPattern("$REDIS_STATE_KEY_PREFIX*") - keys.forEach { key -> - val jobId = key.removePrefix(REDIS_STATE_KEY_PREFIX).toLongOrNull() ?: return@forEach - val redisHash = getRedisHashForJob(jobId) - val allCompleted = redisHash.readAllValues().all { state -> state.status.completed } - if (allCompleted) { - redisHash.delete() - redissonClient.getBucket("$REDIS_STATE_INITIALIZED_KEY_PREFIX$jobId").delete() - // Clear local initialization cache to allow re-initialization if jobId is reused - localInitializedJobs.remove(jobId) - // Do NOT remove counters here - they're needed until job status is properly updated - // Counters will be removed in removeJobState when job is finalized + getCachedJobIds().chunked(CLEANUP_BATCH_SIZE).forEach { clearCompletedStates(it) } + } + + private fun clearCompletedStates(jobIds: List) { + val completedJobIds = + readStates(jobIds).mapNotNull { (jobId, values) -> + val allCompleted = values != null && values.all { it.status.completed } + jobId.takeIf { allCompleted } } + if (completedJobIds.isEmpty()) { + return + } + + // Counters stay live until removeJobState finalizes the job; deleting them here corrupts status updates. + val deleteBatch = redissonClient.createBatch() + completedJobIds.forEach { jobId -> + deleteBatch.getMap(stateKey(jobId)).deleteAsync() + deleteBatch.getBucket(initializedKey(jobId)).deleteAsync() + } + deleteBatch.execute() + completedJobIds.forEach { localInitializedJobs.remove(it) } + } + + private fun readStates(jobIds: List): Map?> { + return try { + val readBatch = redissonClient.createBatch() + val futures = + jobIds.associateWith { jobId -> + readBatch.getMap(stateKey(jobId)).readAllValuesAsync() + } + readBatch.execute() + futures.mapValues { it.value.get() } + } catch (e: Exception) { + logger.warn("Pipelined batch-job state read failed; falling back to per-job reads", e) + jobIds.associateWith { readStateOrNull(it) } + } + } + + private fun readStateOrNull(jobId: Long): Collection? { + return try { + getRedisHashForJob(jobId).readAllValues() + } catch (e: Exception) { + logger.warn("Failed to read batch job state for job $jobId during cleanup", e) + null } } @@ -251,9 +280,15 @@ open class RedisBatchJobStateStorage( } private fun getRedisHashForJob(jobId: Long): RMap { - return redissonClient.getMap("$REDIS_STATE_KEY_PREFIX$jobId") + return redissonClient.getMap(stateKey(jobId)) } + private fun stateKey(jobId: Long) = "$REDIS_STATE_KEY_PREFIX$jobId" + + private fun initializedKey(jobId: Long) = "$REDIS_STATE_INITIALIZED_KEY_PREFIX$jobId" + + private fun startedKey(jobId: Long) = "$REDIS_STARTED_KEY_PREFIX$jobId" + /** * Ensures Redis hash is initialized from DB. Uses local in-memory cache first for O(1) check, * then falls back to Redis marker for cross-instance coordination. @@ -270,7 +305,7 @@ open class RedisBatchJobStateStorage( return } // Check Redis marker for cross-instance coordination - val initKey = "$REDIS_STATE_INITIALIZED_KEY_PREFIX$jobId" + val initKey = initializedKey(jobId) if (redissonClient.getBucket(initKey).get() == true) { localInitializedJobs.add(jobId) return @@ -295,14 +330,17 @@ open class RedisBatchJobStateStorage( } } - private fun removeAllCounters(jobId: Long) { - redissonClient.getAtomicLong("$REDIS_RUNNING_COUNT_KEY_PREFIX$jobId").delete() - redissonClient.getAtomicLong("$REDIS_COMPLETED_CHUNKS_COUNT_KEY_PREFIX$jobId").delete() - redissonClient.getAtomicLong("$REDIS_PROGRESS_COUNT_KEY_PREFIX$jobId").delete() - redissonClient.getAtomicLong("$REDIS_SINGLE_CHUNK_PROGRESS_COUNT_KEY_PREFIX$jobId").delete() - redissonClient.getAtomicLong("$REDIS_FAILED_COUNT_KEY_PREFIX$jobId").delete() - redissonClient.getAtomicLong("$REDIS_CANCELLED_COUNT_KEY_PREFIX$jobId").delete() - redissonClient.getAtomicLong("$REDIS_COMMITTED_COUNT_KEY_PREFIX$jobId").delete() + private fun addCounterDeletesToBatch( + batch: RBatch, + jobId: Long, + ) { + batch.getAtomicLong("$REDIS_RUNNING_COUNT_KEY_PREFIX$jobId").deleteAsync() + batch.getAtomicLong("$REDIS_COMPLETED_CHUNKS_COUNT_KEY_PREFIX$jobId").deleteAsync() + batch.getAtomicLong("$REDIS_PROGRESS_COUNT_KEY_PREFIX$jobId").deleteAsync() + batch.getAtomicLong("$REDIS_SINGLE_CHUNK_PROGRESS_COUNT_KEY_PREFIX$jobId").deleteAsync() + batch.getAtomicLong("$REDIS_FAILED_COUNT_KEY_PREFIX$jobId").deleteAsync() + batch.getAtomicLong("$REDIS_CANCELLED_COUNT_KEY_PREFIX$jobId").deleteAsync() + batch.getAtomicLong("$REDIS_COMMITTED_COUNT_KEY_PREFIX$jobId").deleteAsync() } private fun initializeCountersFromState( diff --git a/backend/data/src/test/kotlin/io/tolgee/batch/BatchJobChunkExecutionQueuePerformanceTest.kt b/backend/data/src/test/kotlin/io/tolgee/batch/BatchJobChunkExecutionQueuePerformanceTest.kt index 07f4176fda..0f1ad373dc 100644 --- a/backend/data/src/test/kotlin/io/tolgee/batch/BatchJobChunkExecutionQueuePerformanceTest.kt +++ b/backend/data/src/test/kotlin/io/tolgee/batch/BatchJobChunkExecutionQueuePerformanceTest.kt @@ -1,5 +1,6 @@ package io.tolgee.batch +import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper import io.micrometer.core.instrument.simple.SimpleMeterRegistry import io.tolgee.Metrics import io.tolgee.batch.data.BatchJobType @@ -39,6 +40,7 @@ class BatchJobChunkExecutionQueuePerformanceTest { usingRedisProvider = mock(), redisTemplate = mock(), metrics = Metrics(SimpleMeterRegistry()), + objectMapper = jacksonObjectMapper(), ) // The internal structures live in the companion object (static), clear between tests executionQueue.clear() diff --git a/backend/data/src/test/kotlin/io/tolgee/batch/BatchJobChunkExecutionQueueTest.kt b/backend/data/src/test/kotlin/io/tolgee/batch/BatchJobChunkExecutionQueueTest.kt index 69017d6c1c..f143aafca7 100644 --- a/backend/data/src/test/kotlin/io/tolgee/batch/BatchJobChunkExecutionQueueTest.kt +++ b/backend/data/src/test/kotlin/io/tolgee/batch/BatchJobChunkExecutionQueueTest.kt @@ -1,5 +1,6 @@ package io.tolgee.batch +import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper import io.micrometer.core.instrument.simple.SimpleMeterRegistry import io.tolgee.Metrics import io.tolgee.batch.data.BatchJobType @@ -31,6 +32,7 @@ class BatchJobChunkExecutionQueueTest { usingRedisProvider = mock(), redisTemplate = mock(), metrics = Metrics(SimpleMeterRegistry()), + objectMapper = jacksonObjectMapper(), ) queue.clear() }