feat(server): implement RedisLocker and IoRedisLocker#846
Conversation
… skipped if Redis is unavailable
🦋 Changeset detectedLatest commit: c4c7f70 The changes in this PR will be included in the next version bump. This PR includes changesets to release 1 package
Not sure what this means? Click here to learn what changesets are. Click here if you're a maintainer who wants to add another changeset to this PR |
ℹ️ Recent review info⚙️ Run configurationConfiguration used: Path: .coderabbit.yml Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (6)
🚧 Files skipped from review as they are similar to previous changes (5)
WalkthroughChangesAdds Redis distributed locking
Estimated code review effort: 4 (Complex) | ~45 minutes 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In @.github/workflows/ci.yml:
- Line 71: Update the Redis service image under the workflow service
configuration from redis:latest to a reviewed Redis version pinned by its
immutable SHA-256 digest, preserving the existing service setup.
In `@packages/server/src/lockers/RedisLocker.ts`:
- Around line 136-146: Initialize the internal abort controller in
RedisLocker.lock and IoRedisLocker.lock as already aborted when
stopSignal.aborted is true, while preserving the existing abort event listener
for later cancellation; update both packages/server/src/lockers/RedisLocker.ts
lines 136-146 and packages/server/src/lockers/IoRedisLocker.ts lines 142-152.
- Around line 168-185: Prevent a late acquireLock() winner from retaining an
orphaned lock after timeout or abort: in
packages/server/src/lockers/RedisLocker.ts at lines 168-185, recheck
cancellation after SET and subscription, then unsubscribe, release ownership,
and return without starting the watchdog; apply the equivalent rollback in
packages/server/src/lockers/IoRedisLocker.ts at lines 174-193, including cleanup
of releaseHandlers and acquired before returning.
In `@packages/server/src/test/IoRedisLocker.test.ts`:
- Around line 38-46: Update the before hook around client connection in
packages/server/src/test/IoRedisLocker.test.ts (lines 38-46) to rethrow
connection failures when running in CI and call this.skip() only outside CI;
apply the same behavior to the corresponding setup hook in
packages/server/src/test/RedisLocker.test.ts (lines 41-48).
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yml
Review profile: CHILL
Plan: Pro
Run ID: 48b1cc61-11c1-401f-8c31-89b89002bbde
📒 Files selected for processing (7)
.changeset/spicy-beans-bet.md.github/workflows/ci.ymlpackages/server/src/lockers/IoRedisLocker.tspackages/server/src/lockers/RedisLocker.tspackages/server/src/lockers/index.tspackages/server/src/test/IoRedisLocker.test.tspackages/server/src/test/RedisLocker.test.ts
| async lock(stopSignal: AbortSignal, requestRelease: RequestRelease): Promise<void> { | ||
| const abortController = new AbortController() | ||
| const onAbort = () => { | ||
| abortController.abort() | ||
| } | ||
| stopSignal.addEventListener('abort', onAbort) | ||
|
|
||
| try { | ||
| const lock = await Promise.race([ | ||
| this.waitTimeout(abortController.signal), | ||
| this.acquireLock(this.id, requestRelease, abortController.signal), |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Propagate an already-aborted signal before acquisition begins.
packages/server/src/lockers/RedisLocker.ts#L136-L146: initialize the internal controller as aborted whenstopSignal.abortedis already true.packages/server/src/lockers/IoRedisLocker.ts#L142-L152: apply the same initial abort propagation.
📍 Affects 2 files
packages/server/src/lockers/RedisLocker.ts#L136-L146(this comment)packages/server/src/lockers/IoRedisLocker.ts#L142-L152
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@packages/server/src/lockers/RedisLocker.ts` around lines 136 - 146,
Initialize the internal abort controller in RedisLocker.lock and
IoRedisLocker.lock as already aborted when stopSignal.aborted is true, while
preserving the existing abort event listener for later cancellation; update both
packages/server/src/lockers/RedisLocker.ts lines 136-146 and
packages/server/src/lockers/IoRedisLocker.ts lines 142-152.
| const lock = await this.locker.redis.set(this.locker.key(id), this.token, { | ||
| condition: 'NX', | ||
| expiration: { | ||
| type: 'PX', | ||
| value: this.locker.lockTimeout, | ||
| }, | ||
| }) | ||
|
|
||
| if (lock) { | ||
| const {subscriber} = this.locker | ||
|
|
||
| // Stores the requestRelease callback which later gets invoked once a message from Redis is received. | ||
| this.listener = () => { | ||
| void Promise.resolve(requestRelease()).catch(() => {}) | ||
| } | ||
| await subscriber.subscribe(this.locker.channel(id), this.listener) | ||
|
|
||
| this.startWatchdog(id, requestRelease) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🔴 Critical | 🏗️ Heavy lift
Prevent late Redis acquisition from creating permanently renewed orphan locks.
Both implementations allow the losing acquireLock() branch of Promise.race() to continue after timeout or abort.
packages/server/src/lockers/RedisLocker.ts#L168-L185: recheck cancellation afterSETand subscription, rolling back ownership and subscription before returning.packages/server/src/lockers/IoRedisLocker.ts#L174-L193: perform equivalent rollback, includingreleaseHandlersandacquiredcleanup.
📍 Affects 2 files
packages/server/src/lockers/RedisLocker.ts#L168-L185(this comment)packages/server/src/lockers/IoRedisLocker.ts#L174-L193
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@packages/server/src/lockers/RedisLocker.ts` around lines 168 - 185, Prevent a
late acquireLock() winner from retaining an orphaned lock after timeout or
abort: in packages/server/src/lockers/RedisLocker.ts at lines 168-185, recheck
cancellation after SET and subscription, then unsubscribe, release ownership,
and return without starting the watchdog; apply the equivalent rollback in
packages/server/src/lockers/IoRedisLocker.ts at lines 174-193, including cleanup
of releaseHandlers and acquired before returning.
…ion and throw err if redis unavailable during tests
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
packages/server/src/test/MemoryLocker.test.ts (1)
109-116: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winUse
assert.rejectsto prevent misleading failure messages.If
lock2.lockunexpectedly resolves,assert(false)throws anAssertionError. The genericcatchblock catches thisAssertionError, evaluatese === ERRORS.ERR_LOCK_TIMEOUT(which is false), and throws a newAssertionErrorwith a confusing message:"error returned is not correct AssertionError: lock2 should not have been acquired".Using
assert.rejectsstandardizes the rejection check and prevents swallowing the original failure context.🛠️ Proposed fix
- try { - await lock2.lock(abortController.signal, async () => { - cancel2() - }) - assert(false, 'lock2 should not have been acquired') - } catch (e) { - assert(e === ERRORS.ERR_LOCK_TIMEOUT, `error returned is not correct ${e}`) - } + await assert.rejects( + lock2.lock(abortController.signal, async () => { + cancel2() + }), + (e: any) => { + assert(e === ERRORS.ERR_LOCK_TIMEOUT, `error returned is not correct ${e}`) + return true + } + )🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/server/src/test/MemoryLocker.test.ts` around lines 109 - 116, Replace the manual try/catch and assert(false) around lock2.lock with assert.rejects, asserting that the promise rejects with ERRORS.ERR_LOCK_TIMEOUT. Preserve the existing lock2.lock invocation and cancellation behavior while allowing unexpected resolution or rejection errors to retain their original assertion context.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@packages/server/src/test/MemoryLocker.test.ts`:
- Around line 49-57: Update the test around lock.lock to use assert.rejects,
ensuring the test fails when the promise unexpectedly resolves. Validate the
rejected value against ERRORS.ERR_LOCK_TIMEOUT without treating it as an Error
or accessing e.message, while preserving the existing body and timeout
assertions.
---
Nitpick comments:
In `@packages/server/src/test/MemoryLocker.test.ts`:
- Around line 109-116: Replace the manual try/catch and assert(false) around
lock2.lock with assert.rejects, asserting that the promise rejects with
ERRORS.ERR_LOCK_TIMEOUT. Preserve the existing lock2.lock invocation and
cancellation behavior while allowing unexpected resolution or rejection errors
to retain their original assertion context.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yml
Review profile: CHILL
Plan: Pro
Run ID: 8dd4d9bb-4daf-4fbd-9fb7-9ab363ae556e
📒 Files selected for processing (6)
.github/workflows/ci.ymlpackages/server/src/lockers/IoRedisLocker.tspackages/server/src/lockers/RedisLocker.tspackages/server/src/test/IoRedisLocker.test.tspackages/server/src/test/MemoryLocker.test.tspackages/server/src/test/RedisLocker.test.ts
🚧 Files skipped from review as they are similar to previous changes (5)
- packages/server/src/test/IoRedisLocker.test.ts
- packages/server/src/lockers/IoRedisLocker.ts
- packages/server/src/test/RedisLocker.test.ts
- packages/server/src/lockers/RedisLocker.ts
- .github/workflows/ci.yml
There was a problem hiding this comment.
Caution
Inline review comments failed to post. This is likely due to GitHub's internal server error or limits when posting large numbers of comments. If you are seeing this consistently it is likely a permissions issue. Please check "Moderation" -> "Code review limits" under your organization settings.
Actionable comments posted: 1
🧹 Nitpick comments (1)
packages/server/src/test/MemoryLocker.test.ts (1)
109-116: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winUse
assert.rejectsto prevent misleading failure messages.If
lock2.lockunexpectedly resolves,assert(false)throws anAssertionError. The genericcatchblock catches thisAssertionError, evaluatese === ERRORS.ERR_LOCK_TIMEOUT(which is false), and throws a newAssertionErrorwith a confusing message:"error returned is not correct AssertionError: lock2 should not have been acquired".Using
assert.rejectsstandardizes the rejection check and prevents swallowing the original failure context.🛠️ Proposed fix
- try { - await lock2.lock(abortController.signal, async () => { - cancel2() - }) - assert(false, 'lock2 should not have been acquired') - } catch (e) { - assert(e === ERRORS.ERR_LOCK_TIMEOUT, `error returned is not correct ${e}`) - } + await assert.rejects( + lock2.lock(abortController.signal, async () => { + cancel2() + }), + (e: any) => { + assert(e === ERRORS.ERR_LOCK_TIMEOUT, `error returned is not correct ${e}`) + return true + } + )🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/server/src/test/MemoryLocker.test.ts` around lines 109 - 116, Replace the manual try/catch and assert(false) around lock2.lock with assert.rejects, asserting that the promise rejects with ERRORS.ERR_LOCK_TIMEOUT. Preserve the existing lock2.lock invocation and cancellation behavior while allowing unexpected resolution or rejection errors to retain their original assertion context.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@packages/server/src/test/MemoryLocker.test.ts`:
- Around line 49-57: Update the test around lock.lock to use assert.rejects,
ensuring the test fails when the promise unexpectedly resolves. Validate the
rejected value against ERRORS.ERR_LOCK_TIMEOUT without treating it as an Error
or accessing e.message, while preserving the existing body and timeout
assertions.
---
Nitpick comments:
In `@packages/server/src/test/MemoryLocker.test.ts`:
- Around line 109-116: Replace the manual try/catch and assert(false) around
lock2.lock with assert.rejects, asserting that the promise rejects with
ERRORS.ERR_LOCK_TIMEOUT. Preserve the existing lock2.lock invocation and
cancellation behavior while allowing unexpected resolution or rejection errors
to retain their original assertion context.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yml
Review profile: CHILL
Plan: Pro
Run ID: 8dd4d9bb-4daf-4fbd-9fb7-9ab363ae556e
📒 Files selected for processing (6)
.github/workflows/ci.ymlpackages/server/src/lockers/IoRedisLocker.tspackages/server/src/lockers/RedisLocker.tspackages/server/src/test/IoRedisLocker.test.tspackages/server/src/test/MemoryLocker.test.tspackages/server/src/test/RedisLocker.test.ts
🚧 Files skipped from review as they are similar to previous changes (5)
- packages/server/src/test/IoRedisLocker.test.ts
- packages/server/src/lockers/IoRedisLocker.ts
- packages/server/src/test/RedisLocker.test.ts
- packages/server/src/lockers/RedisLocker.ts
- .github/workflows/ci.yml
🛑 Comments failed to post (1)
packages/server/src/test/MemoryLocker.test.ts (1)
49-57: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Prevent false positives by failing if no error is thrown.
The current
try...catchblock does not fail the test iflock.lock()unexpectedly succeeds, meaning the test could silently pass when it should fail. Additionally, becauseERRORS.ERR_LOCK_TIMEOUTis an object and not anErrorinstance,e.messageevaluates toundefined.Using
assert.rejectsensures the promise is rejected and handles the validation cleanly.🛠️ Proposed fix
- try { - await lock.lock(abortController.signal, async () => { - throw new Error('panic should not be called') - }) - } catch (e) { - assert(!(e instanceof Error), `error returned is not correct ${e.message}`) - assert('body' in e, 'body is not present in the error') - assert(e.body === ERRORS.ERR_LOCK_TIMEOUT.body) - } + await assert.rejects( + lock.lock(abortController.signal, async () => { + throw new Error('panic should not be called') + }), + (e: any) => { + assert(!(e instanceof Error), 'error returned is not correct') + assert('body' in e, 'body is not present in the error') + assert(e.body === ERRORS.ERR_LOCK_TIMEOUT.body) + return true + } + )📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.await assert.rejects( lock.lock(abortController.signal, async () => { throw new Error('panic should not be called') }), (e: any) => { assert(!(e instanceof Error), 'error returned is not correct') assert('body' in e, 'body is not present in the error') assert(e.body === ERRORS.ERR_LOCK_TIMEOUT.body) return true } )🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/server/src/test/MemoryLocker.test.ts` around lines 49 - 57, Update the test around lock.lock to use assert.rejects, ensuring the test fails when the promise unexpectedly resolves. Validate the rejected value against ERRORS.ERR_LOCK_TIMEOUT without treating it as an Error or accessing e.message, while preserving the existing body and timeout assertions.
|
yo im not even maintaining memory locker |
Murderlon
left a comment
Murderlon
left a comment
There was a problem hiding this comment.
Thermo-nuclear maintainability pass: the API direction is useful, but the current implementation duplicates a non-trivial distributed-lock state machine and its tests. That has already required the same race fix in two places and leaves failure-atomicity gaps around subscription cleanup. I think this needs structural consolidation before merge; the inline comments describe a concrete shape.
| /** | ||
| * @author Oleg Mykula <oleg.mukula@gmail.com> | ||
| */ | ||
| export class IoRedisLocker implements Locker { |
There was a problem hiding this comment.
This is almost a second full copy of RedisLocker: the Lua scripts, timeout/abort race, retry recursion, ownership rollback, watchdog, key/channel naming, and unlock flow are duplicated; only the client command shapes and subscription bookkeeping differ. We already had to apply the late-acquisition fix in both files, which is exactly the drift risk this structure creates.
Please extract one client-agnostic lock engine behind a small adapter (tryAcquire, publish, subscribe/cleanup, compareAndDelete, compareAndExtend) and keep RedisLocker/IoRedisLocker as thin public wrappers. A single deadline/abort-aware acquisition loop could also remove the losing Promise.race branch—and the rollback complexity it created—instead of preserving it twice. This is the code-judo move here: delete most of these ~300 lines rather than maintain two distributed state machines.
| this.listener = () => { | ||
| void Promise.resolve(requestRelease()).catch(() => {}) | ||
| } | ||
| await subscriber.subscribe(this.locker.channel(id), this.listener) |
There was a problem hiding this comment.
The ownership transition is not failure-atomic. Once SET NX succeeds, subscribe() can reject and control exits before the abort rollback below, leaving the key owned until its TTL. The mirror image exists in unlock(): unsubscribe() is awaited before the compare-and-delete, so a subscription cleanup failure prevents releasing ownership after the watchdog has already been stopped. I reproduced both paths with a failing subscriber stub; the release EVAL was called zero times.
Please make compare-and-delete a best-effort finally invariant for both failed acquisition setup and unlock, and add tests for subscribe/unsubscribe rejection. This lifecycle should live once in the shared engine so the two clients cannot diverge.
|
|
||
| const REDIS_URL = process.env.REDIS_URL ?? 'redis://127.0.0.1:6379' | ||
|
|
||
| describe('IoRedisLocker', () => { |
There was a problem hiding this comment.
These 174 lines duplicate the RedisLocker behavioral suite nearly verbatim, so every new invariant and regression test must be remembered twice. Please extract a shared locker contract suite that accepts an async factory/cleanup fixture and run it for both adapters (and for MemoryLocker where the contract applies); keep only client-specific wiring tests here. That would make the implementations genuinely interchangeable and ensure fixes such as abort propagation and failed-subscription rollback are exercised consistently.
Thanks for a thorough review, I'll work on it |
2 participants
This PR adds to new lockers to use which expands the choice from a sole MemoryLocker to also include RedisLocker and IoRedisLocker for better coordination of exclusive upload access across horizontally scaled servers that use tus
What is new
@redis/client) and IoRedisLocker (viaioredis)How does it work
The key is stored on a Redis DB indicating that a locker with a specific ID is currently in use. The value of the key is UUID token generated on a process, it is needed to determine whether a process B has took over the upload that process A was working on, hence took over the Redis entry and changed its value with a new UUID. A Lua script was created in order to execute a few operations at Redis simultaneously without risking any lock requests in-between calls. The script checks whether the current upload ID (if exists) matches the UUID generated by process A. If it was not, but process A still tries to call
DELorSET NXfunctions - it will be denied because the upload no longer belongs to process A. In order to let the process A know that process B wants to take over withrequestRelease, a global subscription pool exists on theLockerinstance (if using IoRedisLocker), which utilizes Redis' Sub/Pub capabilities in order to invoke therequestReleaseon process A, to which process A subscribes and to which process B publishes its nudge. TherequestReleasefunction itself is stored in a hashmap on a mainLockerobject with keys being the channel ids, to which processes subscribe and publish messages. In case with regular RedisLocker instance though, the package provides capabilities to only listen to messages on a specific channel, therefore it lacks the global subscription pool and the hashmap, since it can be stored on the Lock object. Everything else works just as like.Usage
Tests / CI