feat: migrate from Lettuce to Redisson for Redis interactions #15
Conversation
There was a problem hiding this comment.
Pull request overview
This PR migrates the Redis client implementation from Lettuce to Redisson, modernizing connection management and adapting all Redis-backed features (event bus, request-response, cache, and sync structures) to use Redisson's reactive API.
Key Changes:
- Replaced Lettuce client with Redisson for connection management and pub/sub operations
- Updated all sync structures (list, map, set, value) to use Redisson's reactive Bucket and AtomicLong APIs
- Migrated event bus and request-response bus to Redisson's reactive topic API
- Refactored SimpleRedisCache to use Redisson's LocalCachedMap with Caffeine provider
Reviewed changes
Copilot reviewed 10 out of 11 changed files in this pull request and generated 18 comments.
Show a summary per file
| File | Description |
|---|---|
| build.gradle.kts | Replaced Lettuce dependency with Redisson 4.0.0 |
| src/main/kotlin/dev/slne/surf/redis/RedisApi.kt | Migrated from Lettuce client to Redisson, updated connection management and health check APIs |
| src/main/kotlin/dev/slne/surf/redis/event/RedisEventBus.kt | Replaced Lettuce pub/sub listener with Redisson reactive topic subscription |
| src/main/kotlin/dev/slne/surf/redis/request/RequestResponseBus.kt | Updated to use Redisson reactive topics for request/response channels |
| src/main/kotlin/dev/slne/surf/redis/cache/SimpleRedisCache.kt | Migrated to Redisson's LocalCachedMap with Caffeine caching |
| src/main/kotlin/dev/slne/surf/redis/sync/SyncStructure.kt | Updated base class to use Redisson reactive topic for pub/sub |
| src/main/kotlin/dev/slne/surf/redis/sync/value/SyncValue.kt | Migrated to Redisson reactive Bucket and AtomicLong for snapshot/version storage |
| src/main/kotlin/dev/slne/surf/redis/sync/set/SyncSet.kt | Migrated to Redisson reactive Bucket and AtomicLong for snapshot/version storage |
| src/main/kotlin/dev/slne/surf/redis/sync/map/SyncMap.kt | Migrated to Redisson reactive Bucket and AtomicLong for snapshot/version storage |
| src/main/kotlin/dev/slne/surf/redis/sync/list/SyncList.kt | Migrated to Redisson reactive Bucket and AtomicLong for snapshot/version storage |
| src/main/kotlin/dev/slne/surf/redis/util/util.kt | Added custom Mono extension functions for converting to Deferred |
|
@twisti-dev I've opened a new pull request, #16, to work on those changes. Once the pull request is ready, I'll request review from you. |
|
@codex update all kdocs from Lettuce mentions and references to redisson. Also ensure the docs are still correct and correct the invalid ones |
|
Summary
Testing
|
…r improved lazy initialization
|
@codex update all kdocs from Lettuce mentions and references to redisson. Also ensure the docs are still correct and correct the invalid ones |
|
@codex review |
|
Summary
Testing
|
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
surf-redis-core/src/main/kotlin/dev/slne/surf/redis/cache/SimpleRedisCacheImpl.kt
Show resolved
Hide resolved
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 48 out of 61 changed files in this pull request and generated 2 comments.
Comments suppressed due to low confidence (9)
surf-redis-core/src/main/kotlin/dev/slne/surf/redis/request/RequestResponseBusImpl.kt:77
- The logging message replacement is too aggressive. Using
.replace("{", "[").replace("}", "]")on the entire message string can alter legitimate content. This should either replace only curly braces within structured log arguments or use a more targeted approach to avoid corrupting the actual message content.
surf-redis-api/src/main/kotlin/dev/slne/surf/redis/sync/value/SyncValue.kt:121 - Calling
.subscribe()on aMonowithout storing the returnedDisposablemeans there's no way to cancel or clean up this asynchronous operation. Consider storing the disposable and disposing of it during cleanup, or use.block()if synchronous behavior is intended.
surf-redis-api/src/main/kotlin/dev/slne/surf/redis/sync/set/SyncSet.kt:120 - Calling
.subscribe()without storing the returnedDisposablemeans there's no way to cancel or clean up this asynchronous operation. This pattern appears multiple times in sync structures. Consider storing disposables for proper cleanup, or use a different approach if fire-and-forget behavior is intended.
surf-redis-api/src/main/kotlin/dev/slne/surf/redis/sync/set/SyncSet.kt:136 - Calling
.subscribe()without storing the returnedDisposablemeans there's no way to cancel or clean up this asynchronous operation. Consider storing disposables for proper cleanup.
surf-redis-api/src/main/kotlin/dev/slne/surf/redis/sync/set/SyncSet.kt:174 - Calling
.subscribe()on aFluxwithout storing the returnedDisposablecreates a resource leak. The subscription should be stored and disposed of during cleanup to prevent memory leaks and ensure proper resource management.
surf-redis-api/src/main/kotlin/dev/slne/surf/redis/sync/set/SyncSet.kt:195 - Calling
.subscribe()without storing the returnedDisposablecreates a resource leak. Consider storing disposables for proper cleanup.
surf-redis-core/src/main/kotlin/dev/slne/surf/redis/util/KotlinSerializerCache.kt:7 - The visibility of
KotlinSerializerCachewas changed frominternaltopublic. This exposes an internal implementation detail. If this class is not intended to be part of the public API, it should remaininternal.
surf-redis-api/src/main/kotlin/dev/slne/surf/redis/RedisApi.kt:129 - The
Configobject is mutated using.applybut the mutated object is not captured. The setter methods likesetPassword(),setTransportMode(), etc. return theConfiginstance for chaining, but this pattern may not work as expected. Verify that these setter methods actually mutate the object in-place rather than returning a new instance.
surf-redis-api/src/main/kotlin/dev/slne/surf/redis/RedisApi.kt:122 - The password is set using
setPassword()which may not correctly handle the password fromRedisURI. TheRedisURI.getPassword()method likely returns achar[]orString, but it's being used directly. Verify that the RedissonConfig.setPassword()method signature matches the type being passed.
| log.atWarning() | ||
| .log( | ||
| "Received malformed cache invalidation message on topic $invalidationTopicName: ${ | ||
| message.toString().replace("{", "[").replace("}", "]") |
There was a problem hiding this comment.
The same curly brace replacement pattern appears here. This is overly aggressive and can corrupt legitimate content in log messages. Consider a more targeted approach or escaping only when necessary for the logging framework being used.
| api.redissonReactive | ||
| .getBucket<String>(redisKey, StringCodec) | ||
| .expire(ttl.toJavaDuration()) | ||
| .doOnError { e -> | ||
| log.atWarning() | ||
| .withCause(e) | ||
| .log("Failed to refresh TTL for $redisKey") | ||
| } | ||
| .subscribe() |
There was a problem hiding this comment.
The method refreshTtl() calls .subscribe() without storing the Disposable. This creates a fire-and-forget operation that cannot be cancelled or tracked. Since this is called frequently on cache hits, ensure this is the intended behavior and won't cause resource leaks.
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 48 out of 62 changed files in this pull request and generated 3 comments.
Comments suppressed due to low confidence (6)
surf-redis-core/src/main/kotlin/dev/slne/surf/redis/util/KotlinSerializerCache.kt:7
- The visibility of KotlinSerializerCache has been changed from internal to public without justification. This class appears to be an implementation detail and exposing it publicly may create maintenance issues if internal implementation needs to change. Consider whether this class truly needs to be part of the public API or if it should remain internal.
surf-redis-core/src/main/kotlin/dev/slne/surf/redis/request/RequestResponseBusImpl.kt:424 - The RequestResponseBus implementation no longer cleans up request and response type registries in the close method. The old Lettuce-based implementation included clearing requestHandlers, requestTypeRegistry, and responseTypeRegistry. While the new implementation adds these clear operations, it's important to verify that all internal state is properly reset. The current implementation looks correct.
surf-redis-core/src/main/kotlin/dev/slne/surf/redis/event/RedisEventBusImpl.kt:72 - Missing error handling for subscription failures. The setupSubscription method in RedisEventBusImpl subscribes to messages but doesn't handle potential subscription failures. While there is error handling within the stream (onErrorContinue), if the initial subscription itself fails, there's no recovery mechanism. Consider adding a doOnError handler at the subscription level or documenting the expected behavior when subscription fails.
surf-redis-api/src/main/kotlin/dev/slne/surf/redis/sync/SyncStructure.kt:87 - The SyncStructure base class logs errors using "SyncSet" hardcoded in the message. This error message should use the actual structure type or a generic message since SyncStructure is the base class for multiple structure types (SyncSet, SyncList, SyncMap, SyncValue).
surf-redis-api/src/main/kotlin/dev/slne/surf/redis/RedisApi.kt:155 - The RedisApi.create method with pluginDataPath is deprecated but the parameters are ignored in the replacement implementation. The method calls create(serializerModule) which uses RedisCredentialsProvider to get the URI, completely ignoring the pluginDataPath and pluginsPath parameters. This could be confusing for users migrating from the old API, as they might expect the paths to still influence behavior during the deprecation period.
surf-redis-api/src/main/kotlin/dev/slne/surf/redis/sync/SyncStructure.kt:82 - Reactive subscription not properly disposed. The SyncStructure.setupSubscription returns a Flux that is used in init() via subscribe(), but the resulting Disposable is not captured or managed. This means the subscription cannot be properly cleaned up when the structure is closed, potentially causing resource leaks.
| val password = config.password | ||
| if (!password.isNullOrEmpty()) { | ||
| append(URLEncoder.encode(password, StandardCharsets.UTF_8)) | ||
| append('@') | ||
| } |
There was a problem hiding this comment.
Password handling in RedisConfig allows null password, but RedisCredentialsProviderImpl uses URLEncoder.encode without null-checking first. While the code checks for isNullOrEmpty(), it should be consistent with the nullable type declaration. The current implementation is correct but consider documenting why empty string passwords are treated differently from null.
| #org.gradle.configureondemand=true | ||
|
|
||
| version=1.0.0-SNAPSHOT | ||
| version=1.21.11-1.0.0-SNAPSHOT |
There was a problem hiding this comment.
The version string format in gradle.properties has changed from "1.0.0-SNAPSHOT" to "1.21.11-1.0.0-SNAPSHOT". This unusual version format with two version numbers separated by a dash is unconventional and may cause issues with dependency management tools. Semantic versioning typically follows major.minor.patch[-prerelease], not version-version-prerelease.
| version=1.21.11-1.0.0-SNAPSHOT | |
| version=1.21.11-SNAPSHOT |
| invalidationTopic.publish("$nodeId$MESSAGE_DELIMITER$localKey") | ||
| .awaitSingle() |
There was a problem hiding this comment.
Missing return statement in putNull method. The invalidation topic publish call is not awaited, which means the method might complete before the invalidation message is sent to other nodes. This could lead to inconsistent cache state across nodes.
Co-authored-by: TheBjoRedCraft <143264463+TheBjoRedCraft@users.noreply.github.com>
…ve-issue Fix SyncSet.removeIf() failing to replicate across distributed nodes
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 48 out of 62 changed files in this pull request and generated 3 comments.
Comments suppressed due to low confidence (12)
surf-redis-api/src/main/kotlin/dev/slne/surf/redis/sync/SyncStructure.kt:91
- The method
init0()returnsMono<Void>but is defined as an open method with default empty implementation. The naming is unclear - consider renaming to something more descriptive likeonBeforeSubscribe()oronInit()to clarify its purpose as an extension point.
surf-redis-api/src/main/kotlin/dev/slne/surf/redis/sync/SyncStructure.kt:118 - The
.retry()operator on line 118 will retry indefinitely without any backoff or limit. If there's a persistent error in message processing, this could lead to a hot loop consuming resources. Consider using.retryWhen()with exponential backoff or a maximum retry count.
surf-redis-api/src/main/kotlin/dev/slne/surf/redis/sync/value/SyncValue.kt:121 - Same issue as in other sync structures: calls to
publishLocalDelta(...).subscribe()are fire-and-forget without error handling. If publishing fails, local and remote state will diverge. Add error handling via.doOnError()to log failures.
surf-redis-api/src/main/kotlin/dev/slne/surf/redis/sync/set/SyncSet.kt:171 - The method handler invocations in
removeIfare performed synchronously via.subscribe()without storing the disposable. This means if the operation fails, errors are silently ignored. Consider using.doOnError()to log failures or handle them appropriately.
surf-redis-api/src/main/kotlin/dev/slne/surf/redis/sync/set/SyncSet.kt:193 - Multiple calls to
publishLocalDelta(...).subscribe()are made without storing disposables or handling errors. If publishing fails, the failure will be silently ignored and local/remote state will diverge. Consider adding error handling with.doOnError()or storing disposables for proper lifecycle management.
surf-redis-api/src/main/kotlin/dev/slne/surf/redis/sync/map/SyncMap.kt:176 - Same issue as in other sync structures: calls to
publishLocalDelta(...).subscribe()are fire-and-forget without error handling. If publishing fails, local and remote state will diverge. Add error handling via.doOnError()to log failures.
surf-redis-core/src/main/kotlin/dev/slne/surf/redis/event/RedisEventBusImpl.kt:76 - The subscription field is declared as
lateinit varbut is immediately initialized ininit()which is called during API connect. Ifclose()is called beforeinit(), this will throwUninitializedPropertyAccessException. Consider using a nullable type or checking initialization state before disposing.
surf-redis-core/src/main/kotlin/dev/slne/surf/redis/request/RequestResponseBusImpl.kt:414 - The
requestDisposableandresponseDisposablefields are declared aslateinit varbutclose()may be called beforeinit(), resulting inUninitializedPropertyAccessException. Consider using nullable types or checking initialization state before disposing.
surf-redis-api/src/main/kotlin/dev/slne/surf/redis/sync/list/SyncList.kt:217 - Same issue as in other sync structures: calls to
publishLocalDelta(...).subscribe()are fire-and-forget without error handling. If publishing fails, local and remote state will diverge. Add error handling via.doOnError()to log failures.
surf-redis-api/src/main/kotlin/dev/slne/surf/redis/RedisApi.kt:203 - The disposables from
structure.init()are added to a list but never checked for completion status. If a structure fails to initialize, the disposable remains in the list but may be in a failed/completed state. Consider checking disposable state or handling init failures more explicitly.
surf-redis-api/src/main/kotlin/dev/slne/surf/redis/sync/SyncStructure.kt:87 - The error log message incorrectly refers to 'SyncSet' but should be generic since this is in the base
SyncStructureclass which is used by List, Map, Set, and Value. Update the message to use$idor a generic term like 'SyncStructure'.
surf-redis-core/src/main/kotlin/dev/slne/surf/redis/util/KotlinSerializerCache.kt:7 - The
KotlinSerializerCacheclass is being changed frominternalto public visibility. This appears to be unintentional as the class is an implementation detail and should remain internal. Making it public increases the API surface unnecessarily.
| invalidationTopic.publish("$nodeId$MESSAGE_DELIMITER$localKey").awaitSingle() | ||
| } | ||
|
|
||
| override suspend fun cachedOrLoad(key: K, loader: suspend () -> V): V { | ||
| getCached(key)?.let { return it } | ||
| val loaded = loader() | ||
| put(key, loaded) | ||
| return loaded | ||
| } | ||
|
|
||
| override suspend fun cachedOrLoadNullable( | ||
| key: K, | ||
| cacheNull: Boolean, | ||
| loader: suspend () -> V? | ||
| ): V? { | ||
| val existing = getCached(key) | ||
| if (existing != null) return existing | ||
| if (nearCache.getIfPresent(localKey(key)) == CacheEntry.Null) return null | ||
|
|
||
| val loaded = loader() | ||
| when { | ||
| loaded != null -> put(key, loaded) | ||
| cacheNull -> putNull(key) | ||
| } | ||
| return loaded | ||
| } | ||
|
|
||
| private suspend fun putNull(key: K) { | ||
| ensureSubscribed() | ||
| val localKey = localKey(key) | ||
| val redisKey = redisKey(key) | ||
|
|
||
| api.redissonReactive.getBucket<String>(redisKey, StringCodec) | ||
| .set(NULL_MARKER, ttl.toJavaDuration()) | ||
| .awaitSingleOrNull() | ||
|
|
||
| nearCache.put(localKey, CacheEntry.Null) | ||
| invalidationTopic.publish("$nodeId$MESSAGE_DELIMITER$localKey") | ||
| .awaitSingle() | ||
| } | ||
|
|
||
| private fun refreshTtl(localKey: String, redisKey: String) { | ||
| val inserted = refreshGate.asMap().putIfAbsent(localKey, Unit) == null | ||
| if (!inserted) return | ||
|
|
||
| api.redissonReactive | ||
| .getBucket<String>(redisKey, StringCodec) | ||
| .expire(ttl.toJavaDuration()) | ||
| .doOnError { e -> | ||
| log.atWarning() | ||
| .withCause(e) | ||
| .log("Failed to refresh TTL for $redisKey") | ||
| } | ||
| .subscribe() | ||
| } | ||
|
|
||
| override suspend fun invalidate(key: K): Long { | ||
| ensureSubscribed() | ||
|
|
||
| val localKey = localKey(key) | ||
| val redisKey = redisKey(key) | ||
|
|
||
| val deleted = api.redissonReactive | ||
| .getBucket<String>(redisKey, StringCodec) | ||
| .delete() | ||
| .awaitSingle() | ||
|
|
||
| nearCache.invalidate(localKey) | ||
| refreshGate.invalidate(localKey) | ||
| invalidationTopic.publish("$nodeId$MESSAGE_DELIMITER$localKey").awaitSingle() |
There was a problem hiding this comment.
The invalidation topic publish calls at lines 148, 185, and 217 use .awaitSingle() but do not handle errors. If the publish operation fails, it will throw an exception that propagates to the caller. Consider wrapping in try-catch or using .onErrorResume() to handle publication failures gracefully.
| init { | ||
| plugin = this | ||
| RedisInstance.instance.load() |
There was a problem hiding this comment.
The method calls RedisInstance.instance.load() in the init block. If the RedisInstance service is not available or fails to load, this will throw an exception during instance construction before the plugin is fully initialized. Consider deferring this call or wrapping in error handling.
| override fun bootstrap(context: BootstrapContext) { | ||
| dataPath = context.dataDirectory | ||
|
|
||
| RedisInstance.instance.load() |
There was a problem hiding this comment.
The RedisInstance.instance.load() is called in the bootstrap phase. If this throws an exception, the error may not be handled gracefully. Consider adding try-catch with appropriate error logging for failed Redis initialization.
…s, requests, and responses
This pull request introduces a major refactor of the
surf-redislibrary, replacing the Lettuce-based implementation with Redisson and updating the project structure to support a multi-module Gradle build. The changes also remove the old configuration system and update documentation to reflect the new architecture and APIs.Migration from Lettuce to Redisson and Project Modularization:
RedisApiand related classes. The API now usesRedissonClientandRedissonReactiveClientinstead of Lettuce'sRedisClientand connections. (surf-redis-api/src/main/kotlin/dev/slne/surf/redis/RedisApi.kt, [1] [2] [3] [4] [5] [6] [7]README.mdto reference Redisson instead of Lettuce, and clarified threading model references throughout. (README.md, [1] [2] [3] [4] [5] [6] [7] [8]Configuration and Dependency Management:
GlobalConfig.kt,LocalConfig.kt,InternalConfig.kt) in favor of a new credentials provider and simplified configuration, since plugin path-based config is no longer relevant. [1] [2] [3]surf-redis-api,surf-redis-paper,surf-redis-velocity, andsurf-redis-core. Updatedsettings.gradle.ktsand added a newbuild.gradle.ktsforsurf-redis-apiwith updated dependencies for Redisson and test containers. [1] [2]Deprecations and API Changes:
RedisApi.createmethod, since the library is no longer shaded into plugins and does not require plugin-specific paths. (surf-redis-api/src/main/kotlin/dev/slne/surf/redis/RedisApi.kt, surf-redis-api/src/main/kotlin/dev/slne/surf/redis/RedisApi.ktR143-R168)surf-redis-api/src/main/kotlin/dev/slne/surf/redis/RedisApi.kt, surf-redis-api/src/main/kotlin/dev/slne/surf/redis/RedisApi.ktR143-R168)These changes modernize the library, improve maintainability, and align the implementation with current best practices for Kotlin and Redis integration.