perf: reduce Redis round-trips in pub/sub and batch-job state (#3822)

## What

Two independent Redis efficiency improvements, no behavior change:

### 1. Reuse the singleton `ObjectMapper` for pub/sub serialization
`RedisWebsocketEventPublisher`, `BatchJobChunkExecutionQueue` and
`BatchJobCancellationManager` each constructed a fresh
`jacksonObjectMapper()` on every publish — on the websocket and batch
hot paths. They now use the injected singleton, matching how
`RedisPubSubReceiver` already consumes it.

### 2. Pipeline batch-job state cleanup with Redisson `RBatch`
`RedisBatchJobStateStorage` issued one Redis round-trip per key:

- **`removeJobState`** deleted the state hash, 7 counters and 2 marker
buckets as ~10 sequential calls → now one `RBatch`.
- **`clearUnusedStates`** read each job's hash and deleted it one at a
time → now reads every job's state in one pipelined batch and deletes
the completed ones in a second. Work is **chunked** (100 jobs/pass) so
peak pipeline-response size and materialized state stay bounded
regardless of job count. When a pipelined decode fails (e.g. a
codec/enum change makes a value unreadable), it **falls back to isolated
per-job reads** so one bad job can't abort cleanup of the rest.

## Tests

`BatchJobChunkExecutionQueueTest` / `...PerformanceTest` updated for the
new constructor arg.

New `RedisBatchJobStateStorageTest` (integration, real Redis) covers the
parts with no prior coverage:
- `removeJobState` pipelines its deletes into a single `RBatch` (spy
asserts `createBatch` once, never per-key `getAtomicLong`)
- `clearUnusedStates` clears completed jobs across multiple chunks while
preserving counters, the started marker and unfinished jobs
- `clearUnusedStates` skips a job whose state cannot be read and still
cleans the healthy ones

`clearUnusedStates` previously had **no** test coverage (it is only
driven by a `@Scheduled` cleaner); `removeJobState`'s hash-clearing
remains covered end-to-end by `BatchJobsGeneralWithRedisTest`.

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

- **Improvements**
- Improved Redis-backed batch job cleanup and removal while preserving
active job information and progress counters.
  - Enhanced reliability when individual job states cannot be read.
  - Improved processing of larger sets of completed jobs.
- Standardized serialization for Redis-backed batch jobs and WebSocket
events for more consistent operation.
- **Tests**
- Added coverage for cleanup, batched deletion, chunk boundaries,
partial read failures, and preservation of unfinished jobs.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: bdshadow

backend/api/src/main/kotlin/io/tolgee/websocket/RedisWebsocketEventPublisher.kt

@@ -1,16 +1,17 @@
 package io.tolgee.websocket
 
 import org.springframework.data.redis.core.StringRedisTemplate
-import tools.jackson.module.kotlin.jacksonObjectMapper
+import tools.jackson.databind.ObjectMapper
 
 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,

backend/api/src/main/kotlin/io/tolgee/websocket/WebsocketPublisherConfiguration.kt

@@ -6,6 +6,7 @@ import org.springframework.context.annotation.Bean
 import org.springframework.context.annotation.Configuration
 import org.springframework.data.redis.core.StringRedisTemplate
 import org.springframework.messaging.simp.SimpMessagingTemplate
+import tools.jackson.databind.ObjectMapper
 
 @Configuration
 class WebsocketPublisherConfiguration(
@@ -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))
   }

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<ConfigurableApplicationContext> {
+      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<Long>()
+
+  @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<String>())
+
+    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<String, String>("$STATE_KEY_PREFIX$corruptJobId", StringCodec.INSTANCE)["1"] = "corrupt"
+    redissonClient.getBucket<Boolean>("$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<Boolean>("$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<Boolean>("$STATE_INITIALIZED_PREFIX$jobId").isExists
+
+  private fun isStartedBucketPresent(jobId: Long) = redissonClient.getBucket<Boolean>("$STARTED_PREFIX$jobId").isExists
+}

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

@@ -20,7 +20,7 @@ import org.springframework.data.redis.core.StringRedisTemplate
 import org.springframework.stereotype.Component
 import org.springframework.transaction.PlatformTransactionManager
 import org.springframework.transaction.annotation.Transactional
-import tools.jackson.module.kotlin.jacksonObjectMapper
+import tools.jackson.databind.ObjectMapper
 
 @Component
 class BatchJobCancellationManager(
@@ -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)

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

@@ -23,7 +23,7 @@ import org.springframework.data.redis.core.StringRedisTemplate
 import org.springframework.scheduling.annotation.Scheduled
 import org.springframework.stereotype.Component
 import org.springframework.transaction.annotation.Transactional
-import tools.jackson.module.kotlin.jacksonObjectMapper
+import tools.jackson.databind.ObjectMapper
 import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.ConcurrentLinkedDeque
 import java.util.concurrent.atomic.AtomicInteger
@@ -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