fix(streamer): await sendToManySubscriptions async dispatch#38681
fix(streamer): await sendToManySubscriptions async dispatch#38681Shreyas2004wagh wants to merge 8 commits intoRocketChat:developfrom
Conversation
|
|
Looks like this PR is not ready to merge, because of the following issues:
Please fix the issues and try again If you have any trouble, please check the PR guidelines |
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
WalkthroughReworked Streamer.sendToManySubscriptions to perform per-subscription async permission checks and sends concurrently via Promise.allSettled, skip the origin subscription, and log per-subscription permission or delivery failures while continuing delivery to other subscribers. Changes
Sequence Diagram(s)sequenceDiagram
participant Streamer as Streamer
participant SubA as Subscription A
participant SubB as Subscription B
participant SubC as Subscription C
participant Logger as SystemLogger
Streamer->>SubA: isEmitAllowed(subA) (async)
Streamer->>SubB: isEmitAllowed(subB) (async)
Streamer->>SubC: isEmitAllowed(subC) (async)
Note over SubA,SubC: permission checks resolve/reject independently
alt allowed
SubA-->>Streamer: allowed
Streamer->>SubA: send(msg)
SubA-->>Streamer: send fulfilled / rejected
else denied
SubB-->>Streamer: denied (skip send)
end
alt isEmitAllowed rejects
SubC-->>Streamer: error
Streamer->>Logger: log permission error (eventName, streamName)
end
Streamer->>Logger: log any per-subscription send failures from allSettled
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~20 minutes Poem
🚥 Pre-merge checks | ✅ 3✅ Passed checks (3 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. 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 |
There was a problem hiding this comment.
Pull request overview
This PR fixes a critical async handling bug in the streamer module where forEach(async ...) was not awaiting async operations, causing unreliable message delivery and potentially dropped errors.
Changes:
- Replaced
forEach(async ...)withPromise.allSettled()to properly await all subscription deliveries - Added explicit error handling and logging for failed subscription deliveries
- Ensures async permission checks complete deterministically before the method resolves
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## develop #38681 +/- ##
===========================================
+ Coverage 70.49% 70.60% +0.10%
===========================================
Files 3175 3189 +14
Lines 111094 112705 +1611
Branches 20045 20402 +357
===========================================
+ Hits 78311 79570 +1259
- Misses 30738 31077 +339
- Partials 2045 2058 +13
Flags with carried forward coverage won't be shown. Click here to find out more. 🚀 New features to boost your workflow:
|
|
Codecov is flagging the new async branches introduced in I’m adding focused unit tests to cover the updated behavior (async per-subscription handling, error logging, and delivery semantics) and will push them shortly to address patch coverage. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Fix all issues with AI agents
In `@apps/meteor/tests/unit/server/modules/streamer/streamer.module.spec.ts`:
- Around line 55-58: The afterEach cleanup uses
createdStreamers.splice(0).forEach((name) => delete
StreamerCentral.instances[name]) which returns a boolean from the forEach
callback and triggers Biome lint; change the cleanup to iterate with a for...of
loop (or a simple for loop) over createdStreamers.splice(0) and perform delete
StreamerCentral.instances[name] inside the loop so the callback does not return
a value; update the afterEach that also calls sinon.restore() accordingly.
🧹 Nitpick comments (1)
apps/meteor/tests/unit/server/modules/streamer/streamer.module.spec.ts (1)
7-19: Nit: Remove inline comments per coding guidelines.The
// no-op for unit testcomments on the stub methods can be dropped — the empty body already conveys the intent. As per coding guidelines,**/*.{ts,tsx,js}: "Avoid code comments in the implementation."Suggested diff
class TestStreamer extends Streamer<any> { - registerPublication(): void { - // no-op for unit test - } + registerPublication(): void {} - registerMethod(): void { - // no-op for unit test - } + registerMethod(): void {} changedPayload(): string { return 'payload'; } }
📜 Review details
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
apps/meteor/tests/unit/server/modules/streamer/streamer.module.spec.ts
🧰 Additional context used
📓 Path-based instructions (2)
**/*.{ts,tsx,js}
📄 CodeRabbit inference engine (.cursor/rules/playwright.mdc)
**/*.{ts,tsx,js}: Write concise, technical TypeScript/JavaScript with accurate typing in Playwright tests
Avoid code comments in the implementation
Files:
apps/meteor/tests/unit/server/modules/streamer/streamer.module.spec.ts
**/*.spec.ts
📄 CodeRabbit inference engine (.cursor/rules/playwright.mdc)
**/*.spec.ts: Use descriptive test names that clearly communicate expected behavior in Playwright tests
Use.spec.tsextension for test files (e.g.,login.spec.ts)
Files:
apps/meteor/tests/unit/server/modules/streamer/streamer.module.spec.ts
🧠 Learnings (9)
📚 Learning: 2025-11-24T17:08:17.065Z
Learnt from: CR
Repo: RocketChat/Rocket.Chat PR: 0
File: .cursor/rules/playwright.mdc:0-0
Timestamp: 2025-11-24T17:08:17.065Z
Learning: Applies to apps/meteor/tests/e2e/**/*.spec.ts : Ensure tests run reliably in parallel without shared state conflicts
Applied to files:
apps/meteor/tests/unit/server/modules/streamer/streamer.module.spec.ts
📚 Learning: 2025-11-24T17:08:17.065Z
Learnt from: CR
Repo: RocketChat/Rocket.Chat PR: 0
File: .cursor/rules/playwright.mdc:0-0
Timestamp: 2025-11-24T17:08:17.065Z
Learning: Applies to apps/meteor/tests/e2e/**/*.spec.ts : Group related tests in the same file
Applied to files:
apps/meteor/tests/unit/server/modules/streamer/streamer.module.spec.ts
📚 Learning: 2025-11-24T17:08:17.065Z
Learnt from: CR
Repo: RocketChat/Rocket.Chat PR: 0
File: .cursor/rules/playwright.mdc:0-0
Timestamp: 2025-11-24T17:08:17.065Z
Learning: Applies to apps/meteor/tests/e2e/**/*.spec.ts : Maintain test isolation between test cases in Playwright tests
Applied to files:
apps/meteor/tests/unit/server/modules/streamer/streamer.module.spec.ts
📚 Learning: 2025-11-24T17:08:17.065Z
Learnt from: CR
Repo: RocketChat/Rocket.Chat PR: 0
File: .cursor/rules/playwright.mdc:0-0
Timestamp: 2025-11-24T17:08:17.065Z
Learning: Applies to apps/meteor/tests/e2e/**/*.spec.ts : All test files must be created in `apps/meteor/tests/e2e/` directory
Applied to files:
apps/meteor/tests/unit/server/modules/streamer/streamer.module.spec.ts
📚 Learning: 2025-12-10T21:00:54.909Z
Learnt from: KevLehman
Repo: RocketChat/Rocket.Chat PR: 37091
File: ee/packages/abac/jest.config.ts:4-7
Timestamp: 2025-12-10T21:00:54.909Z
Learning: Rocket.Chat monorepo: Jest testMatch pattern '<rootDir>/src/**/*.spec.(ts|js|mjs)' is valid in this repo and used across multiple packages (e.g., packages/tools, ee/packages/omnichannel-services). Do not flag it as invalid in future reviews.
Applied to files:
apps/meteor/tests/unit/server/modules/streamer/streamer.module.spec.ts
📚 Learning: 2025-11-24T17:08:17.065Z
Learnt from: CR
Repo: RocketChat/Rocket.Chat PR: 0
File: .cursor/rules/playwright.mdc:0-0
Timestamp: 2025-11-24T17:08:17.065Z
Learning: Applies to apps/meteor/tests/e2e/**/*.spec.ts : Utilize Playwright fixtures (`test`, `page`, `expect`) for consistency in test files
Applied to files:
apps/meteor/tests/unit/server/modules/streamer/streamer.module.spec.ts
📚 Learning: 2025-11-24T17:08:17.065Z
Learnt from: CR
Repo: RocketChat/Rocket.Chat PR: 0
File: .cursor/rules/playwright.mdc:0-0
Timestamp: 2025-11-24T17:08:17.065Z
Learning: Applies to apps/meteor/tests/e2e/**/*.spec.ts : Use `test.step()` for complex test scenarios to improve organization in Playwright tests
Applied to files:
apps/meteor/tests/unit/server/modules/streamer/streamer.module.spec.ts
📚 Learning: 2025-11-24T17:08:17.065Z
Learnt from: CR
Repo: RocketChat/Rocket.Chat PR: 0
File: .cursor/rules/playwright.mdc:0-0
Timestamp: 2025-11-24T17:08:17.065Z
Learning: Applies to apps/meteor/tests/e2e/**/*.spec.ts : Use `expect` matchers for assertions (`toEqual`, `toContain`, `toBeTruthy`, `toHaveLength`, etc.) instead of `assert` statements in Playwright tests
Applied to files:
apps/meteor/tests/unit/server/modules/streamer/streamer.module.spec.ts
📚 Learning: 2025-11-24T17:08:17.065Z
Learnt from: CR
Repo: RocketChat/Rocket.Chat PR: 0
File: .cursor/rules/playwright.mdc:0-0
Timestamp: 2025-11-24T17:08:17.065Z
Learning: Applies to apps/meteor/tests/e2e/**/*.spec.ts : Ensure clean state for each test execution in Playwright tests
Applied to files:
apps/meteor/tests/unit/server/modules/streamer/streamer.module.spec.ts
🪛 Biome (2.3.14)
apps/meteor/tests/unit/server/modules/streamer/streamer.module.spec.ts
[error] 57-57: This callback passed to forEach() iterable method should not return a value.
Either remove this return or remove the returned value.
(lint/suspicious/useIterableCallbackReturn)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (3)
- GitHub Check: 📦 Build Packages
- GitHub Check: cubic · AI code reviewer
- GitHub Check: CodeQL-Build
🔇 Additional comments (3)
apps/meteor/tests/unit/server/modules/streamer/streamer.module.spec.ts (3)
60-80: LGTM!The gate-based pattern cleanly validates that dispatch is deferred until the async permission check resolves. Well structured.
82-104: LGTM!Origin filtering, allowed dispatch, and denied-subscription skipping are all well covered. The test correctly relies on
Setinsertion-order iteration for deterministic stub sequencing.
106-130: LGTM!Good resilience test — confirms that a rejected permission check is logged with the expected structured payload and does not block delivery to other subscribers.
✏️ Tip: You can disable this entire section by setting review_details to false in your review settings.
|
Added focused unit tests covering async dispatch and error handling in |
|
I reverted all changes outside the streamer scope. After reviewing CI failures:
Given the streamer-only scope of this PR, there is no appropriate cross-layer fix to apply here. |
There was a problem hiding this comment.
Actionable comments posted: 1
📜 Review details
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (2)
apps/meteor/server/modules/streamer/streamer.module.tsapps/meteor/tests/unit/server/modules/streamer/streamer.module.spec.ts
🚧 Files skipped from review as they are similar to previous changes (1)
- apps/meteor/tests/unit/server/modules/streamer/streamer.module.spec.ts
🧰 Additional context used
📓 Path-based instructions (1)
**/*.{ts,tsx,js}
📄 CodeRabbit inference engine (.cursor/rules/playwright.mdc)
**/*.{ts,tsx,js}: Write concise, technical TypeScript/JavaScript with accurate typing in Playwright tests
Avoid code comments in the implementation
Files:
apps/meteor/server/modules/streamer/streamer.module.ts
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (3)
- GitHub Check: 📦 Build Packages
- GitHub Check: cubic · AI code reviewer
- GitHub Check: CodeQL-Build
🔇 Additional comments (1)
apps/meteor/server/modules/streamer/streamer.module.ts (1)
293-316: Core concurrent-dispatch fix is correct; notePromise.allvsPromise.allSettledin PR description.The use of
await Promise.all([...subscriptions].map(async ...))correctly awaits all per-subscription async permission checks and sends concurrently, with one failure not blocking others (enforced by the internal try/catch). TheretransmitToSelfguard, truthyallowedcheck, and optional-chainedsocket?.send()are all preserved correctly.Minor note: the PR description references
Promise.allSettled, but since the entire callback body is wrapped in try/catch, no mapped promise can reject —Promise.allandPromise.allSettledare functionally equivalent in this shape. No change needed, but the PR description should be updated for accuracy.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@apps/meteor/server/modules/streamer/streamer.module.ts`:
- Around line 307-313: The catch in the streamer delivery path currently calls
SystemLogger.debug which silences real failures; update the catch in the
delivery block inside streamer.module.ts to log at a higher level (use
SystemLogger.error per PR intent, or at minimum SystemLogger.warn), keeping the
same structured fields (msg, eventName, streamName: this.name) and include the
caught err details so failures from isEmitAllowed or the TransformMessage/getMsg
callback are visible in production; adjust the log call where the catch
currently invokes SystemLogger.debug to SystemLogger.error and ensure err is
passed through in the log payload.
apps/meteor/tests/unit/server/modules/streamer/streamer.module.spec.ts
Outdated
Show resolved
Hide resolved
| let release!: (value: boolean) => void; | ||
| const gate = new Promise<boolean>((resolve) => { | ||
| release = resolve; | ||
| }); | ||
|
|
||
| sinon.stub(streamer, 'isEmitAllowed').returns(gate as any); | ||
|
|
||
| const sendPromise = streamer.sendToManySubscriptions(new Set([sub.entry]), undefined, 'event', [], 'test-msg'); | ||
|
|
||
| await Promise.resolve(); | ||
| expect(sub.send.called).to.equal(false); | ||
|
|
||
| release(true); | ||
| await sendPromise; |
There was a problem hiding this comment.
can u simplify these test? this part is a bit confusing, i'm sure there's some work that can be done diff
2 participants
Proposed changes (including videos or screenshots)
This PR fixes async dispatch handling in the streamer delivery path.
forEach(async ...)insendToManySubscriptionswithPromise.allSettled(...)over mapped async tasks.SystemLogger.error) for failed subscription deliveries.void this.sendToManySubscriptions(...)) unchanged, so this remains a scoped reliability fix with no API contract changes.Issue(s)
Steps to test or reproduce
allowEmitis async (for example, existing asyncallowEmitrules like__my_messages__).Further comments
apps/meteor/server/modules/streamer/streamer.module.ts.Summary by CodeRabbit
Bug Fixes
Tests