Give copies of Properties their own header table - #96
Merged
Conversation
Tracing was attached to a producer by deriving from it. `TracingProducerImpl` overrode every publishing entry point, both `send` overloads and `trySend`, with three near identical bodies which copied the message, called `createAndTag` and re-wrapped the confirmation callback. That shape has to be extended by hand for every new overload, and had already failed that way once. The `send` overload taking an explicit mandatory flag silently bypassed tracing until it was noticed and a third copy of the body added in bloomberg#91. Inheritance was doing very little work here. The subclass carried three data members and one behaviour, and that behaviour is not a variation on being a producer, it is a step in publishing a message. Use composition instead. `ProducerImpl` now optionally holds an `rmqp::ProducerTagger`, invoked once per send from `prepareMessageForSending`. `rmqa::TracingTagger` adapts the configured `rmqp::ProducerTracing` onto it, and `TracingProducerImpl` is deleted along with its factory. The exchange name is passed to the tagger rather than held by it, so a tagger has no per producer state and one instance serves every producer on a connection. `sendImpl` and `trySend` share `prepareMessageForSending` because the step cannot move into `doSend`. It has to run before the wait on the outstanding confirm limit, and those two differ in precisely what sits between the two, a blocking wait against a try. The helper gives that ordering one home. This is the pattern the consumer already uses. `TracingConsumerImpl` is an empty shell around a factory which builds a plain `ConsumerImpl` with a different `MessageGuard::Factory`, so consumer tracing has always been injected as a collaborator. The producer was the odd one out. Only the empty shell class is left over on that side, and it can go whenever the consumer is next touched. The `rmqp::ProducerTracing` interface is unchanged, so tracing implementations need no change. Three behaviours they rely on which the interface cannot express are preserved. 1. The hook runs on the calling thread 2. It runs before the wait on the outstanding confirm limit 3. The context it returns lives until the broker responds
`rmqt::Properties` holds its headers as a `bsl::shared_ptr<rmqt::FieldTable>`,
and neither `Properties` nor `Message` declared a copy constructor, so copying
either one copied the pointer and not the table. Every copy of a message
therefore aliased one `FieldTable`, which is a `bsl::map` with no locking.
Sending is asynchronous. `send()` posts the publish to the event loop and
returns, and the headers are walked much later, on the event loop thread, to
build the content header frame. A caller which reuses a `Properties` across
sends is then mutating a map which is concurrently being read.
```
rmqt::Properties props = setupProps();
for (int x = 0; x < 10; ++x) {
(*props.headers)["appheader"] = x;
producer.send(message, routingKey, callback);
}
```
Nothing about tracing is required for this. It applies equally to the
transformer path, which emplaces its marker headers into the same shared
table, and to a plain send with no hooks at all. A tracer setting properties
on every send is simply the most likely writer to make it visible.
Fix it in `Properties` rather than at the call sites which happen to matter
today. The copy constructor and assignment operator now give the copy its own
header table, so the copies the producer and consumer already make are correct
without either of them doing anything special. A destructor is declared
alongside them for consistency.
One minor observable change. A caller which reads a header back out of its
own table after `send()`, a trace id say, no longer sees it. That only ever
worked by virtue of this aliasing, and only for a single threaded sender.
`Methods_BasicProperties.Headers` asserted that `setProperties` left the
stored headers pointer equal to the one passed in. That is the aliasing being
removed, so it now compares the tables by value.
This costs copies, which is accepted for now and left to be addressed with
other publishing performance work. Measured on the send path, four
`Properties` copies per send each deep copy the header table where previously
they shared a pointer, and that is a lower bound because the tests exercise a
mock event loop which skips the real handler copy. Declaring the copy
operations also suppresses the implicit move operations on compilers which
have them, so a move of a `Properties` or a `Message` is now a deep copy too.
Both are worth revisiting when C++03 support can be dropped and move
operations can be declared.
`PropertiesTests` covers the copy and assignment behaviour directly, including
that every field is copied, that a null header table stays null, and that self
assignment is safe. `SendDoesNotShareHeaderTableWithCaller` runs against the
plain and the tracing producer and fails against both without the fix, and
`TracingDoesNotMutateCallerHeaders` drives a hook which injects in place, as
the real ones do, and checks the injected header reaches the broker but not
the caller.
willhoy
approved these changes
Sep 9, 2026
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
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.
Problem statement
This fixes an issue with
rmqt::Properties, which holds its headers as absl::shared_ptr<rmqt::FieldTable>, not having it's own copy before enqueuing a send or modifying them for tracing.It's split into two commits - the first some refactoring to improve message tracing, making the fix simpler.
Proposed changes
Tag outgoing messages via composition
Tracing was attached to a producer by deriving from it.
TracingProducerImploverrode every publishing entry point, both
sendoverloads andtrySend,with three near identical bodies which copied the message, called
createAndTagand re-wrapped the confirmation callback. That shape has to beextended by hand for every new overload, and had already failed that way once.
The
sendoverload taking an explicit mandatory flag silently bypassedtracing until it was noticed and a third copy of the body added in
#91.
Inheritance was doing very little work here. The subclass carried three data
members and one behaviour, and that behaviour is not a variation on being a
producer, it is a step in publishing a message. Use composition instead.
ProducerImplnow optionally holds anrmqp::ProducerTagger, invoked onceper send from
prepareMessageForSending.rmqa::TracingTaggeradapts theconfigured
rmqp::ProducerTracingonto it, andTracingProducerImplisdeleted along with its factory. The exchange name is passed to the tagger
rather than held by it, so a tagger has no per producer state and one instance
serves every producer on a connection.
sendImplandtrySendshareprepareMessageForSendingbecause the stepcannot move into
doSend. It has to run before the wait on the outstandingconfirm limit, and those two differ in precisely what sits between the two, a
blocking wait against a try. The helper gives that ordering one home.
This is the pattern the consumer already uses.
TracingConsumerImplis anempty shell around a factory which builds a plain
ConsumerImplwith adifferent
MessageGuard::Factory, so consumer tracing has always beeninjected as a collaborator. The producer was the odd one out. Only the empty
shell class is left over on that side, and it can go whenever the consumer is
next touched.
The
rmqp::ProducerTracinginterface is unchanged, so tracing implementationsneed no change. Three behaviours they rely on which the interface cannot
express are preserved.
Give copies of Properties their own header table
rmqt::Propertiesholds its headers as absl::shared_ptr<rmqt::FieldTable>,and neither
PropertiesnorMessagedeclared a copy constructor, so copyingeither one copied the pointer and not the table. Every copy of a message
therefore aliased one
FieldTable, which is absl::mapwith no locking.Sending is asynchronous.
send()posts the publish to the event loop andreturns, and the headers are walked much later, on the event loop thread, to
build the content header frame. A caller which reuses a
Propertiesacrosssends is then mutating a map which is concurrently being read.
Nothing about tracing is required for this. It applies equally to the
transformer path, which emplaces its marker headers into the same shared
table, and to a plain send with no hooks at all. A tracer setting properties
on every send is simply the most likely writer to make it visible.
Fix it in
Propertiesrather than at the call sites which happen to mattertoday. The copy constructor and assignment operator now give the copy its own
header table, so the copies the producer and consumer already make are correct
without either of them doing anything special. A destructor is declared
alongside them for consistency.
One minor observable change. A caller which reads a header back out of its
own table after
send(), a trace id say, no longer sees it. That only everworked by virtue of this aliasing, and only for a single threaded sender.
Methods_BasicProperties.Headersasserted thatsetPropertiesleft thestored headers pointer equal to the one passed in. That is the aliasing being
removed, so it now compares the tables by value.
This costs copies, which is accepted for now and left to be addressed with
other publishing performance work. Measured on the send path, four
Propertiescopies per send each deep copy the header table where previouslythey shared a pointer, and that is a lower bound because the tests exercise a
mock event loop which skips the real handler copy. Declaring the copy
operations also suppresses the implicit move operations on compilers which
have them, so a move of a
Propertiesor aMessageis now a deep copy too.Both are worth revisiting when C++03 support can be dropped and move
operations can be declared.
PropertiesTestscovers the copy and assignment behaviour directly, includingthat every field is copied, that a null header table stays null, and that self
assignment is safe.
SendDoesNotShareHeaderTableWithCallerruns against theplain and the tracing producer and fails against both without the fix, and
TracingDoesNotMutateCallerHeadersdrives a hook which injects in place, asthe real ones do, and checks the injected header reaches the broker but not
the caller.