fix(websocket): deliver events to every subscriber of an instance - #181
Open
prakash-dev-code wants to merge 1 commit into
Open
Conversation
The websocket producer kept a single connection per instance in `clients map[string]*websocket.Conn`. Registering a second subscriber silently replaced the first, and `RemoveClient` deleted the whole instance entry when either one disconnected. Consuming events in two browser tabs therefore delivered to only the most recent one, and closing that tab cut delivery for the remaining subscriber too. Changes: - `clients` becomes `map[string][]*client`. Deregistration now matches the specific connection instead of removing the instance entry, so sibling subscribers keep receiving events. - Each connection carries its own write mutex. Events are dispatched from independent goroutines (the `go CallWebhook` calls in the whatsmeow event handler) and gorilla/websocket permits only one concurrent writer per connection, so two simultaneous events could interleave writes on the same socket. - `Produce` snapshots the subscriber list under the read lock and performs writes outside it, so a slow or half-open socket no longer blocks deliveries for other instances. - Failed writes prune the dead connection, and `Produce` now returns nil. `sendToQueueOrWebhook` aborts the remaining producers when one returns an error, so a closed browser tab previously suppressed RabbitMQ, NATS and webhook delivery for that same event. Verified with two concurrent subscribers on one instance: both receive identical event streams, and disconnecting one leaves the other unaffected.
Reviewer's GuideThe websocket producer now delivers each event to every subscriber, safely serializes concurrent writes per socket, performs network I/O outside the registry lock, and removes failed connections without propagating websocket errors that would suppress other producers. Sequence diagram for websocket event fan-outsequenceDiagram
participant Producer as websocketProducer
participant Registry as Subscriber registry
participant ClientA as Subscriber A
participant ClientB as Subscriber B
participant Broadcast as Broadcast subscriber
Producer->>Registry: RLock and snapshot subscribers
Registry-->>Producer: instance clients and broadcast clients
Producer->>ClientA: writeJSON(message)
Producer->>ClientB: writeJSON(message)
Producer->>Broadcast: writeJSON(message)
Producer->>Registry: Lock and prune failed clients
Producer-->>Producer: return nil
File-Level Changes
Possibly linked issues
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 1 issue
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="pkg/events/websocket/websocket_producer.go" line_range="174" />
<code_context>
+
+ var failed []*client
+ for _, c := range targets {
+ if err := c.writeJSON(message); err != nil {
+ p.loggerWrapper.GetLogger(instanceID).LogError(
+ "Erro ao enviar mensagem websocket para %s: %v", instanceID, err)
+ failed = append(failed, c)
}
- p.loggerWrapper.GetLogger(instanceID).LogInfo("Mensagem websocket enviada com sucesso para instância %s na fila %s", instanceID, queueName)
}
- // Envia para todos os clientes broadcast
- for _, conn := range p.broadcast {
- err := conn.WriteJSON(message)
- if err != nil {
- p.loggerWrapper.GetLogger(instanceID).LogError("Erro ao enviar mensagem broadcast websocket: %v", err)
- continue
+ if len(failed) > 0 {
+ p.clientsMux.Lock()
+ for _, c := range failed {
+ if remaining := drop(p.clients[instanceID], c); len(remaining) == 0 {
+ delete(p.clients, instanceID)
+ } else {
+ p.clients[instanceID] = remaining
+ }
+ p.broadcast = drop(p.broadcast, c)
}
+ p.clientsMux.Unlock()
</code_context>
<issue_to_address>
**issue (bug_risk):** When `writeJSON` fails, the connection is removed from the producer's subscriber lists but is never closed here. Its read goroutine can remain blocked in `ReadMessage`, leaving the failed socket and goroutine allocated indefinitely when the peer does not close cleanly.
**Triggers:** When a socket becomes unusable during a write while its peer remains half-open.
**Suggested fix:** Close each failed connection after pruning it, for example by calling `_ = c.conn.Close()` so the read loop exits and performs its normal cleanup.
```suggestion
p.broadcast = drop(p.broadcast, c)
_ = c.conn.Close()
```
</issue_to_address>Sourcery assessment
Approval pending. 1 finding to address first.
Blocking findings: pkg/events/websocket/websocket_producer.go:174
Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| } else { | ||
| p.clients[instanceID] = remaining | ||
| } | ||
| p.broadcast = drop(p.broadcast, c) |
There was a problem hiding this comment.
issue (bug_risk): When writeJSON fails, the connection is removed from the producer's subscriber lists but is never closed here. Its read goroutine can remain blocked in ReadMessage, leaving the failed socket and goroutine allocated indefinitely when the peer does not close cleanly.
Triggers: When a socket becomes unusable during a write while its peer remains half-open.
Suggested fix: Close each failed connection after pruning it, for example by calling _ = c.conn.Close() so the read loop exits and performs its normal cleanup.
Suggested change
| p.broadcast = drop(p.broadcast, c) | |
| p.broadcast = drop(p.broadcast, c) | |
| _ = c.conn.Close() |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
The websocket producer stored a single connection per instance in
clients map[string]*websocket.Conn. Registering a second subscriber silentlyreplaced the first, and
RemoveClientremoved the whole instance entry wheneither disconnected. Consuming events in two browser tabs delivered to only the
most recent one, and closing that tab cut delivery for the other.
This changes
clientstomap[string][]*clientand matches deregistration tothe specific connection. Three related issues are fixed in the same path:
(the
go CallWebhookcalls in the whatsmeow event handler) andgorilla/websocket permits one concurrent writer per connection, so two
simultaneous events could interleave writes on the same socket. Each
connection now has its own write mutex.
Producenow snapshots subscribers under the readlock and writes outside it, so a slow or half-open socket no longer blocks
deliveries for other instances.
sendToQueueOrWebhookaborts the remainingproducers when one returns an error, so a closed browser tab previously
suppressed RabbitMQ/NATS/webhook delivery for the same event. Failed writes
now prune the dead connection and
Producereturns nil.No public API or behaviour change for existing single-subscriber setups.
Related Issue
N/A — found while building against the websocket producer. Happy to open an
issue first if you'd prefer that.
Type of Change
Testing
Verified with two concurrent websocket subscribers on one instance: both
receive identical event streams, and disconnecting one leaves the other
unaffected.
go build ./...,make fmtandgolangci-lint run ./pkg/events/websocket/...are all clean.Checklist
Summary by Sourcery
Ensure websocket events are delivered reliably to every active subscriber while isolating slow or disconnected connections.
Bug Fixes:
Enhancements: