Fix[MQB]: stop catchup before reaching common iterator #1046

dorjesinpo · 2025-12-11T01:41:10Z

When RelayQueueEngine calls catchUp for a single App, the PushStream may have a gap resulting in the App first element being behind common iterator.
A solution is to introduce a seqeunce number (only for PushStream messages)
For RootQueueEngine, catchUp still can rely on comparing current guid against the common iterator.
Since both engines use the same mqbi::StorageIterator interface, the duality is expressed as bdlb::Variant<bsls::Types::Uint64, bmqt::MessageGUID>

678098 · 2025-12-15T16:14:29Z

src/groups/mqb/mqbblp/mqbblp_queueengineutil.cpp

+    bsls::TimeInterval*                                          delay,
+    mqbi::StorageIterator*                                       reader,
+    mqbi::StorageIterator*                                       start,
+    const bdlb::Variant<bsls::Types::Uint64, bmqt::MessageGUID>& stop)


As discussed with @dorjesinpo, I will try to find a way to provide stop without building a Variant object. Template arg might work well, however, there might be cleaner ways to refactor the code

678098

Counter thing introduces a multi-threading bug

src/groups/mqb/mqbblp/mqbblp_pushstream.cpp

src/groups/mqb/mqbblp/mqbblp_relayqueueengine.t.cpp

src/groups/mqb/mqbblp/mqbblp_pushstream.h

bmq-oss-ci

Build 3170 of commit 75ad47e has completed with FAILURE

src/groups/mqb/mqbblp/mqbblp_pushstream.cpp

678098 · 2025-12-18T19:37:20Z

src/groups/mqb/mqbblp/mqbblp_pushstream.cpp

    // cannot erase the GUID because of the d_iterator

-    if (d_iterator->second.numElements() == 0) {
+    if (message().numElements() == 0) {


message() accessor now checks that !atEnd(), the same check happens at line 124. It means that the same thing is being checked twice during removeCurrentElement call

Correct. We can remove the assert in message() if we want to. Everywhere it is called, there is the same (duplicate) check prior to the call. But, it looks safer to keep the assert. If we want optimization, we can do something about asserts maybe.

const PushStream::Message& message() const is a protected API, we fully control how to use it from within this class. All its usages are visible and easy to review.
I think it's worth to add checks where we use the class from external code but it's not worth to check any action in internal API, if we have trust in object's state due to the checks that were already performed.

678098 · 2025-12-18T19:38:57Z

src/groups/mqb/mqbblp/mqbblp_pushstream.cpp

    clearCache();

-    if (d_iterator->second.numElements() == 0) {
+    if (message().numElements() == 0) {


!atEnd is checked twice now during advance call

678098 · 2025-12-18T19:39:28Z

src/groups/mqb/mqbblp/mqbblp_pushstream.cpp

    if (d_currentOrdinal > appOrdinal) {
        d_currentOrdinal = 0;
-        d_currentElement = d_iterator->second.front();
+        d_currentElement = message().d_appMessages.front();


!atEnd is checked twice now in this function

src/groups/mqb/mqbblp/mqbblp_pushstream.h

678098 · 2025-12-18T19:50:02Z

src/groups/mqb/mqbblp/mqbblp_relayqueueengine.cpp

+    mqbblp::PushStreamIterator* d_start;
+    mqbblp::PushStreamIterator* d_stop;


Suggested change

mqbblp::PushStreamIterator* d_start;

mqbblp::PushStreamIterator* d_stop;

mqbblp::PushStreamIterator* d_start_p;

mqbblp::PushStreamIterator* d_stop_p;

src/groups/mqb/mqbblp/mqbblp_relayqueueengine.cpp

678098 · 2025-12-18T19:54:25Z

src/groups/mqb/mqbblp/mqbblp_rootqueueengine.cpp

+    mqbi::StorageIterator* d_start;
+    mqbi::StorageIterator* d_stop;


Suggested change

mqbi::StorageIterator* d_start;

mqbi::StorageIterator* d_stop;

mqbi::StorageIterator* d_start_p;

mqbi::StorageIterator* d_stop_p;

src/groups/mqb/mqbblp/mqbblp_relayqueueengine.cpp

678098 · 2025-12-18T19:55:15Z

src/groups/mqb/mqbblp/mqbblp_rootqueueengine.cpp

+    , d_stop(stop)
+    , d_doAdvance(false)
+    {
+        // NOTHING


Suggested change

// NOTHING

// PRECONDITIONS

BSLS_ASSERT_SAFE(d_start_p);

BSLS_ASSERT_SAFE(d_end_p);

Signed-off-by: dorjesinpo <[email protected]>

bmq-oss-ci

Build 3175 of commit c080fc3 has completed with FAILURE

678098

A couple more comments

678098 · 2026-01-06T21:29:53Z

src/groups/mqb/mqbblp/mqbblp_pushstream.h

+        /// Return number of Elements in the list
+        unsigned int numElements() const;
+
+        bsls::Types::Uint64 sequenceNumber() const;


Is there a reason to keep data signed and accessor unsigned? Let's keep both unsigned?

678098 · 2026-01-06T21:36:00Z

src/groups/mqb/mqbblp/mqbblp_pushstream.cpp

    // cannot erase the GUID because of the d_iterator

-    if (d_iterator->second.numElements() == 0) {
+    if (message().numElements() == 0) {


const PushStream::Message& message() const is a protected API, we fully control how to use it from within this class. All its usages are visible and easy to review.
I think it's worth to add checks where we use the class from external code but it's not worth to check any action in internal API, if we have trust in object's state due to the checks that were already performed.

678098 · 2026-01-06T21:57:07Z

src/groups/mqb/mqbblp/mqbblp_relayqueueengine.cpp

+        return 0;
+    }
+
+    if (!d_stop_p->atEnd()) {


We only advance d_start_p. This means that the result of d_stop_p->atEnd() never changes unless we pass a pointer to the same mqbi::StorageIterator:

VirtualIterator iter(start_p, start_p);

In this case we will return 0 on line 206 and never reach the check if (!d_stop_p->atEnd()) {

It is possible to compute d_stop_p->sequenceNumber() just once in constructor:

const bsls::Types::Uint64 d_stopSequenceNumber; VirtualIterator(mqbblp::PushStreamIterator* start, mqbblp::PushStreamIterator* stop) : d_start_p(start) , d_stopSequenceNumber((stop && !stop->atEnd()) ? stop->sequenceNumber() : bsl::numeric_limits<bsls::Types::Uint64>::max()) , d_doAdvance(false) { // PRECONDITIONS BSLS_ASSERT_SAFE(d_start_p); }

And simplify the condition:

if (d_stopSequenceNumber <= d_start_p->sequenceNumber()) { return 0; }

Note that when we call next() multiple times, we always check 2 if conditions, after this change we will only check 1.

678098 · 2026-01-06T21:59:27Z

src/groups/mqb/mqbblp/mqbblp_rootqueueengine.cpp

+    }
+
+    if (!d_stop_p->atEnd()) {
+        if (d_start_p->guid() == d_stop_p->guid()) {


Same here. When we call next() multiple times we always check these 2 conditions. We can just cache MessageGuid and check one:

if (d_start_p->guid() == d_stopGuid) { return 0; }

678098 · 2026-01-06T22:00:29Z

src/groups/mqb/mqbblp/mqbblp_rootqueueengine.cpp

+    // this prevents file set from closing possibly for a very long time.
+    // Make sure to invalidate any cached data within this iterator after use.
+    // TODO: refactor iterators to remove cached data.
+    d_start_p->clearCache();


Don't we need to do the same for RelayQueueEngine::VirtualIterator?

dorjesinpo requested a review from a team as a code owner December 11, 2025 01:41

dorjesinpo added the bug Something isn't working label Dec 11, 2025

dorjesinpo assigned pniedzielski Dec 11, 2025

dorjesinpo requested a review from pniedzielski December 11, 2025 17:10

dorjesinpo changed the title ~~stop catchup before reaching common iterator~~ Fix[MQB]: stop catchup before reaching common iterator Dec 11, 2025

678098 self-requested a review December 15, 2025 15:14

678098 reviewed Dec 15, 2025

View reviewed changes

678098 requested changes Dec 15, 2025

View reviewed changes

src/groups/mqb/mqbblp/mqbblp_pushstream.cpp Outdated Show resolved Hide resolved

src/groups/mqb/mqbblp/mqbblp_relayqueueengine.t.cpp Outdated Show resolved Hide resolved

src/groups/mqb/mqbblp/mqbblp_pushstream.h Outdated Show resolved Hide resolved

dorjesinpo force-pushed the fix/stop-catchup branch 3 times, most recently from 77e816f to 75ad47e Compare December 18, 2025 14:46

dorjesinpo assigned 678098 Dec 18, 2025

bmq-oss-ci bot reviewed Dec 18, 2025

View reviewed changes

678098 reviewed Dec 18, 2025

View reviewed changes

dorjesinpo added 2 commits December 31, 2025 11:49

stop catchup before reaching common iterator

cdd2cd7

Signed-off-by: dorjesinpo <[email protected]>

replacing variant with interface

3cedfcc

Signed-off-by: dorjesinpo <[email protected]>

dorjesinpo force-pushed the fix/stop-catchup branch from 75ad47e to 8912eb8 Compare December 31, 2025 17:23

Addressing review

c080fc3

Signed-off-by: dorjesinpo <[email protected]>

dorjesinpo force-pushed the fix/stop-catchup branch from 8912eb8 to c080fc3 Compare December 31, 2025 17:33

bmq-oss-ci bot reviewed Dec 31, 2025

View reviewed changes

678098 self-requested a review January 6, 2026 19:41

678098 reviewed Jan 6, 2026

View reviewed changes

678098 assigned dorjesinpo and unassigned 678098 Jan 6, 2026

		mqbblp::PushStreamIterator* d_start;
		mqbblp::PushStreamIterator* d_stop;

		mqbi::StorageIterator* d_start;
		mqbi::StorageIterator* d_stop;

-        // NOTHING
+        // PRECONDITIONS
+        BSLS_ASSERT_SAFE(d_start_p);
+        BSLS_ASSERT_SAFE(d_end_p);

Fix[MQB]: stop catchup before reaching common iterator #1046

Are you sure you want to change the base?

Fix[MQB]: stop catchup before reaching common iterator #1046

Uh oh!

Conversation

dorjesinpo commented Dec 11, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

678098 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

bmq-oss-ci bot left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

bmq-oss-ci bot left a comment

Choose a reason for hiding this comment

Uh oh!

678098 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants