Only allow a single dev queue consumer to dequeue at once #1737
Conversation
|
WalkthroughThis update adds a method to check the WebSocket’s ready state before sending messages and adjusts the parameters used when instantiating a queue consumer by passing a unique connection identifier. In addition, a Redis client is now integrated into the queue consumer to track connection state changes and enforce active connection consistency. A new getter for Redis options has also been provided in the Pub/Sub utility module. Changes
Sequence Diagram(s)sequenceDiagram
participant C as Client
participant A as AuthenticatedSocketConnection
participant D as DevQueueConsumer
participant R as Redis
C->>A: Initiate message send
A->>A: Execute canSendMessage() check
alt WebSocket is OPEN
A->>D: Instantiate DevQueueConsumer (passing id)
D->>R: Set active connection key
D->>D: Check message-sending capability in doWork()
alt Redis key mismatch or timeout
D->>D: Stop consumer
else
D->>A: Process message tasks
end
else
A-->>C: Block sending (connection not open)
end
Possibly related PRs
Poem
✨ Finishing Touches
Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media? 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (3)
apps/webapp/app/v3/marqs/devQueueConsumer.server.ts (3)
57-58: Consider renaming for clarity
_connectionLostAtcan be renamed (e.g.,_lastConnectionLossTime) to more clearly express how it's used.
68-71: Confirm keyPrefix usage
MergingdevPubSub.redisOptionsunder the same key prefix could collide across multiple dev queue consumers. Consider incorporating the environment-specific identifier into the prefix to safeguard against collisions.
265-280: Consider backoff for repeated checks
Retrying every second can be noisy if the connection remains unavailable for a long period. Implementing an exponential (or configurable) backoff could reduce resource usage while waiting for reconnection.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (3)
apps/webapp/app/v3/authenticatedSocketConnection.server.ts(1 hunks)apps/webapp/app/v3/marqs/devQueueConsumer.server.ts(4 hunks)apps/webapp/app/v3/utils/zodPubSub.server.ts(1 hunks)
⏰ Context from checks skipped due to timeout of 90000ms (5)
- GitHub Check: e2e / 🧪 CLI v3 tests (windows-latest - pnpm)
- GitHub Check: e2e / 🧪 CLI v3 tests (windows-latest - npm)
- GitHub Check: typecheck / typecheck
- GitHub Check: units / 🧪 Unit Tests
- GitHub Check: Analyze (javascript-typescript)
🔇 Additional comments (5)
apps/webapp/app/v3/utils/zodPubSub.server.ts (1)
119-121: Ensure sensitive data is not inadvertently exposed
Returning the entireredisconfiguration may include connection secrets if present. Please review whether exposing these options is necessary or if a redacted or read-only object is more suitable.apps/webapp/app/v3/marqs/devQueueConsumer.server.ts (2)
25-25: Import usage looks good
No issues spotted with importingcreateRedisClientandRedisClienthere.
61-61: Constructor parameter addition
Introducing a separateidparameter is a helpful way to uniquely identify each consumer. Ensure callers consistently provide meaningful or globally unique IDs.apps/webapp/app/v3/authenticatedSocketConnection.server.ts (2)
46-46: Clever readiness check
Usingws.readyState === WebSocket.OPENforcanSendMessageis straightforward. Double-check edge cases aroundCLOSINGorCLOSEDstates to avoid unexpected failures.
49-49: Validate uniqueness of connection ID
Passingthis.idis convenient; confirm if collisions are possible in multi-tenant or multi-connection scenarios. If collisions are a concern, consider prefixing with relevant environment information.
| await this._redisClient.set(`connection:${this.env.id}`, this.id); | ||
|
|
There was a problem hiding this comment.
🛠️ Refactor suggestion
Add TTL or cleanup routine for the Redis key
Setting connection:${this.env.id} may lead to stale entries if the consumer crashes. Consider using an expiration or a cleanup process, so inactive connections won't block new consumers.
| this._connectionLostAt = undefined; | ||
|
|
||
| const currentConnection = await this._redisClient.get(`connection:${this.env.id}`); | ||
|
|
||
| if (currentConnection !== this.id) { | ||
| logger.debug("Another connection is active, stopping the consumer", { | ||
| currentConnection, | ||
| env: this.env, | ||
| }); | ||
|
|
||
| await this.stop("Another connection is active"); | ||
| return; | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Assess concurrency handling
Using Redis to store the active connection ID is a good start, but there may be a race condition if multiple consumers try to claim the same environment concurrently. For robust exclusivity, consider using distributed locks or atomic check-and-set operations.
ericallam
force-pushed
the
ea-branch-18
branch
from
fa6acc6 to
19a5718
Compare
There was a problem hiding this comment.
Caution
Inline review comments failed to post. This is likely due to GitHub's limits when posting large numbers of comments. If you are seeing this consistently, please check "Code review limits" under "Moderation" settings.
Actionable comments posted: 4
🧹 Nitpick comments (3)
packages/core/src/v3/logger/taskLogger.ts (1)
112-114: Method signature should match the interface declaration.While functionally correct, the method signature in NoopTaskLogger doesn't match the interface declaration. For better code clarity and maintainability, it should include the same parameters as defined in the interface.
- startSpan(): Span { + startSpan(name: string, options?: SpanOptions): Span { return {} as Span; }packages/core/src/v3/streams/asyncIterableStream.ts (2)
1-1: Consider clarifying the hybrid type signature.
AsyncIterableStream<T>mergesAsyncIterable<T>withReadableStream<T>. While this approach allows flexible usage, be mindful that some consumers might expect standardReadableStreammethods without implementing async iteration themselves, or vice versa. Consider adding documentation clarifying usage and any limitations.
7-7: Avoid using theanytype fortransformedStream.Using
anydefeats the benefits of static typing. You can leverage the known type structure ofpipeThroughto provide more precise type definitions:- const transformedStream: any = source.pipeThrough(new TransformStream(transformer)); + const transformedStream = source.pipeThrough(new TransformStream<S, T>(transformer));
🛑 Comments failed to post (4)
packages/python/README.md (2)
58-73:
⚠️ Potential issueUndefined variable usage in streaming loop.
It appears that
resultis assigned frompython.stream.runScript(...), but the subsequentfor await (const chunk of streamingResult)references a different variable,streamingResult, which is never defined. This will cause a runtime error. A fix is to use the same variable identifier consistently:- const result = python.stream.runScript("my_script.py", ["hello", "world"]); + const streamingResult = python.stream.runScript("my_script.py", ["hello", "world"]); // result is an async iterable/readable stream - for await (const chunk of streamingResult) { + for await (const chunk of streamingResult) { logger.debug("convert-url-to-markdown", { url: payload.url, chunk, }); }📝 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.export const myStreamingScript = task({ id: "my-streaming-python-script", run: async () => { // You can also stream the output of the script - const result = python.stream.runScript("my_script.py", ["hello", "world"]); + const streamingResult = python.stream.runScript("my_script.py", ["hello", "world"]); // result is an async iterable/readable stream - for await (const chunk of streamingResult) { + for await (const chunk of streamingResult) { logger.debug("convert-url-to-markdown", { url: payload.url, chunk, }); } }, });
66-71:
⚠️ Potential issueMissing definition of 'payload'.
payload.urlis accessed inside the loop, but no definition ofpayloadis visible within this snippet. This will lead to a ReferenceError. Verify or definepayloadbefore it’s used:for await (const chunk of streamingResult) { - logger.debug("convert-url-to-markdown", { - url: payload.url, - chunk, - }); + const url = "some_url_value_or_variable"; + logger.debug("convert-url-to-markdown", { + url, + chunk, + }); }📝 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.for await (const chunk of streamingResult) { const url = "some_url_value_or_variable"; logger.debug("convert-url-to-markdown", { url, chunk, }); }packages/core/src/v3/streams/asyncIterableStream.ts (2)
88-90: 🛠️ Refactor suggestion
Selective cancellation approach might be too specific.
The code checks
if ("kill" in asyncIterable) { (asyncIterable as any).kill(); }, but not all async iterables will have akillmethod. This can introduce fragile assumptions. Consider a more generic abort or cleanup mechanism or properly type-guard to handle process-like objects separately.
12-14:
⚠️ Potential issueReview the hardcoded
IteratorResult<string>return type.The iterator’s
nextmethod specifiesPromise<IteratorResult<string>>, but thevalueread fromreader.read()likely matches the generic typeT(or a transformed version). Consider using a type parameter instead ofstring:- async next(): Promise<IteratorResult<string>> { + async next(): Promise<IteratorResult<T>> {📝 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.async next(): Promise<IteratorResult<T>> { const { done, value } = await reader.read(); return done ? { done: true, value: undefined } : { done: false, value };
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
apps/webapp/app/v3/marqs/devQueueConsumer.server.ts (1)
265-295: Add periodic Redis key refresh.To prevent the Redis key from expiring while the connection is still active, implement a periodic refresh mechanism.
async #enable() { if (this._enabled) { return; } await this._redisClient.set(`connection:${this.env.id}`, this.id, "EX", 60 * 60 * 24); // 24 hours + // Set up periodic refresh of the connection lock + this._refreshInterval = setInterval(async () => { + if (this._enabled) { + await this._redisClient.expire(`connection:${this.env.id}`, 60 * 60 * 24); + } + }, 60 * 60 * 1000); // Refresh every hour this._enabled = true; // Rest of the method... } public async stop(reason: string = "CLI disconnected") { if (!this._enabled) { return; } logger.debug("[DevQueueConsumer] Stopping dev queue consumer", { env: this.env }); this._enabled = false; + + // Clear the refresh interval + if (this._refreshInterval) { + clearInterval(this._refreshInterval); + this._refreshInterval = undefined; + } + + // Release the connection lock + await this._redisClient.del(`connection:${this.env.id}`); // Rest of the method... }Don't forget to add the corresponding property to the class:
private _refreshInterval?: NodeJS.Timeout;
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
apps/webapp/app/v3/marqs/devQueueConsumer.server.ts(4 hunks)
⏰ Context from checks skipped due to timeout of 90000ms (4)
- GitHub Check: e2e / 🧪 CLI v3 tests (windows-latest - pnpm)
- GitHub Check: e2e / 🧪 CLI v3 tests (windows-latest - npm)
- GitHub Check: typecheck / typecheck
- GitHub Check: Analyze (javascript-typescript)
🔇 Additional comments (7)
apps/webapp/app/v3/marqs/devQueueConsumer.server.ts (7)
25-25: Good import addition.Adding the Redis client import is necessary for the new connection tracking functionality.
57-58: Well-structured property definitions.The private properties for connection tracking are appropriately defined. The optional
_connectionLostAtproperty will help detect long disconnections.
61-61: Constructor parameter appropriately added.Adding the ID parameter is essential for uniquely identifying each consumer connection.
68-71: Initialize Redis client with proper configuration.The Redis client initialization looks good, leveraging the existing Redis options from the pubsub module.
246-246: Add TTL or cleanup routine for the Redis key.Setting
connection:${this.env.id}may lead to stale entries if the consumer crashes. Consider using an expiration or a cleanup process, so inactive connections won't block new consumers.The 24-hour expiration is a good start, but consider implementing a periodic refresh of this key while the connection is active, or a cleanup routine when the consumer is explicitly stopped.
265-281: Good connection validation implementation.The implementation correctly checks if messages can be sent and handles connection losses gracefully:
- Tracks connection loss time
- Stops the consumer if disconnected for more than 60 seconds
- Returns early to prevent further processing
- Schedules next check appropriately
Regarding line 281, the
returnstatement is indeed necessary to prevent execution from continuing, addressing Matt's comment.
283-295: Assess concurrency handling.Using Redis to store the active connection ID is a good start, but there may be a race condition if multiple consumers try to claim the same environment concurrently. For robust exclusivity, consider using distributed locks or atomic check-and-set operations.
Additionally, consider implementing a heartbeat mechanism to periodically refresh the Redis key, ensuring that if a consumer becomes unresponsive without proper cleanup, the lock will eventually expire allowing new consumers to connect.
| @@ -235,6 +243,8 @@ export class DevQueueConsumer { | |||
| return; | |||
| } | |||
|
|
|||
| await this._redisClient.set(`connection:${this.env.id}`, this.id, "EX", 60 * 60 * 24); // 24 hours | |||
There was a problem hiding this comment.
🛠️ Refactor suggestion
Add cleanup in the stop method.
When the consumer is explicitly stopped (in the stop() method), you should delete the Redis key to immediately release the connection lock.
public async stop(reason: string = "CLI disconnected") {
if (!this._enabled) {
return;
}
logger.debug("[DevQueueConsumer] Stopping dev queue consumer", { env: this.env });
this._enabled = false;
+
+ // Release the connection lock
+ await this._redisClient.del(`connection:${this.env.id}`);
// Create the session
const session = await disconnectSession(this.env.id);
// Rest of the method...
}Also applies to: 283-295
Summary by CodeRabbit