fix(rabbitmq): serialize access to the shared AMQP channel - #363
Open
roberiocavalcante wants to merge 1 commit into
Open
fix(rabbitmq): serialize access to the shared AMQP channel#363roberiocavalcante wants to merge 1 commit into
roberiocavalcante wants to merge 1 commit into
Conversation
PublishToRabbit used a single package-level amqp091.Channel from several
goroutines at once, with no synchronization. Events are published
concurrently from the per-client event handler (wmiau.go, via safeGo)
and from the send-message handler (handlers.go, via `go
sendToGlobalRabbit`), and both reach the same channel.
An amqp091.Channel is not safe for concurrent use, and the two calls
made on it here are guarded differently by the library:
- Channel.Publish takes the channel's internal mutex (ch.m) and holds
it while sendOpen writes the method frame, the header frame and the
N body frames.
- Channel.QueueDeclare goes through ch.call() and takes no channel
mutex at all. Only the connection-level lock is taken, and that one
is acquired per frame.
Since every publish declares the queue first, a Queue.Declare method
frame can therefore be written in between another goroutine's content
frames on the same channel id. The broker sees a method frame while it
is still assembling content and kills the connection with:
unexpected_frame: "expected content body, got non content body
frame instead"
Every event in flight on that connection is lost. In production this
lost 2184 events across 6 windows in a single day, with the worst burst
(1227 events in 4 minutes) lining up with the day's peak event volume,
which is exactly when the interleaving window is hit most often.
Concurrent QueueDeclare calls are broken for the same reason: two
callers both wait on ch.rpc and can consume each other's
Queue.DeclareOk.
Separately, handleConnectionErrors swapped rabbitConn, rabbitChannel
and rabbitEnabled while publishers were reading them, so a publisher
could observe rabbitEnabled==true together with the channel of an
already dead connection.
Fix: add rabbitMu, held across declare+publish so the frames of one
message cannot interleave with another call, and used for every read
and write of the connection state. The dead connection is now cleared
on disconnect so no publisher can reach its channel while reconnection
is in progress.
QueueDeclare is deliberately left on the publish path: it is what makes
a queue deleted at runtime come back, and dropping or memoizing it
would turn that case into silent message loss, which is the failure
this commit is fixing.
Adds rabbitmq_test.go, covering the state race under `go test -race`.
Verified by removing the lock from each of the three functions in turn;
each removal reproduces the data race and fails the test.
roberiocavalcante
pushed a commit
to devari-tecnologia/wuzapi
that referenced
this pull request
Aug 31, 2026
This fork's default state is byte-identical to upstream. PATCHES.md makes any divergence explicit and auditable: what is patched, why, the upstream issue and PR it is tied to, and the date it gets re-decided if there is no outcome. The single entry is the RabbitMQ concurrent-publish fix (asternic#363), carried only while that PR is open.
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.
Fixes #362.
PublishToRabbituses a single package-levelamqp091.Channelfrom several goroutines with no synchronization.Channel.Publishholds the channel mutexch.macross its method/header/body frames, butChannel.QueueDeclaregoes throughch.call()and takes no channel mutex — and every publish declares the queue first. AQueue.Declareframe can therefore be written between another goroutine's header and body frames, and the broker closes the connection withunexpected_frame: "expected content body, got non content body frame instead". Every event in flight is lost, andrabbitEnabledstaysfalseafterwards.What changes
rabbitMuguardsrabbitConn/rabbitChannel/rabbitEnabledand is held across theQueueDeclare+Publishpair, so one message's frames cannot interleave with another call.setRabbitConnection/rabbitIsEnabledaccessors, so the reconnection goroutine and the publishers never observe half-updated state. The dead connection is now cleared on disconnect, instead of only flippingrabbitEnabled.rabbitmq_test.gocovering the state races undergo test -race.QueueDeclareis deliberately left on the publish path. It is what makes a queue deleted at runtime come back; without it, a publish to a missing queue withmandatory=falseis discarded with no error at all — which would trade a loud failure for a silent one.Verification
go build ./...,go vet ./...,go test -race ./...,gofmt,go mod tidy— all clean.The test was falsified rather than just written: removing the lock from each of the three functions in turn reproduces a data race and fails the test.
setRabbitConnectionrabbitIsEnabledPublishToRabbit-racecovers the shared-state races but not the frame interleaving, which happens on the wire rather than in memory; that part is argued from the library source referenced in #362.Happy to reshape this if you'd rather declare the queues once at startup, or use one channel per publisher.