From 9f84295bcd26cc34cb38aefa8338e44d892d7564 Mon Sep 17 00:00:00 2001 From: Christopher Beard Date: Wed, 7 Oct 2026 10:44:48 -0400 Subject: [PATCH 1/5] UT[mqbs]: proxy VS removal must not release refs of unpushed Apps On a proxy, a message's refCount counts only the Apps it was pushed to, but every App slot of a new DataStreamMessage starts as e_PUT, which counts as pending. Removing the virtual storage of an App that never received the message still releases a reference, so a later CONFIRM from one App reaches zero references while another App holds the message. This test reproduces that bug in the storage layer with minimal step. Signed-off-by: Christopher Beard --- .../mqb/mqbs/mqbs_inmemorystorage.t.cpp | 70 +++++++++++++++++++ 1 file changed, 70 insertions(+) diff --git a/src/groups/mqb/mqbs/mqbs_inmemorystorage.t.cpp b/src/groups/mqb/mqbs/mqbs_inmemorystorage.t.cpp index 979e531802..b34bd2a0d4 100644 --- a/src/groups/mqb/mqbs/mqbs_inmemorystorage.t.cpp +++ b/src/groups/mqb/mqbs/mqbs_inmemorystorage.t.cpp @@ -44,6 +44,7 @@ #include #include #include +#include #include #include #include @@ -87,6 +88,7 @@ using namespace bsl; // - capacityMeter_limitMessages // capacityMeter_limitBytes // - garbageCollect +// - proxyRemoveVirtualStorage_refCountOfUnpushedApp // - addQueueOpRecordHandle //----------------------------------------------------------------------------- @@ -1434,6 +1436,74 @@ BMQTST_TEST(garbageCollect) BMQTST_ASSERT_EQ(storage.numBytes(mqbu::StorageKey::k_NULL_KEY), 0); } +BMQTST_TEST(proxyRemoveVirtualStorage_refCountOfUnpushedApp) +// ------------------------------------------------------------------------ +// Proxy has 3 Apps. A PUSH arrives for app1 and app2 only (refCount 2), +// as done by 'RelayQueueEngine::storePushIfProxy'. Removing app3's +// virtual storage must not release a reference app3 never held. +// +// Every App slot of a new 'DataStreamMessage' starts as 'e_PUT', which +// counts as pending, so app3's untouched slot looks like it holds a ref. +// ------------------------------------------------------------------------ +{ + bmqtst::TestHelper::printTestName( + "proxyRemoveVirtualStorage_refCountOfUnpushedApp"); + + bmqu::MemOutStream errDescription(bmqtst::TestHelperUtil::allocator()); + + Tester tester(k_PROXY_PARTITION_ID, bmqtst::TestHelperUtil::allocator()); + tester.configure(k_INT64_MAX, k_INT64_MAX); + + mqbs::ReplicatedStorage& storage = tester.storage(); + + storage.addVirtualStorage(errDescription, k_APP_ID1, k_APP_KEY1); + storage.addVirtualStorage(errDescription, k_APP_ID2, k_APP_KEY2); + storage.addVirtualStorage(errDescription, k_APP_ID3, k_APP_KEY3); + + bdlbb::PooledBlobBufferFactory bufferFactory( + 1024, + bmqtst::TestHelperUtil::allocator()); + const bmqt::MessageGUID guid = generateRandomGUID(); + const int data = 0; + const bsl::shared_ptr appData( + new (*bmqtst::TestHelperUtil::allocator()) + bdlbb::Blob(&bufferFactory, bmqtst::TestHelperUtil::allocator()), + bmqtst::TestHelperUtil::allocator()); + bdlbb::BlobUtil::append(appData.get(), + reinterpret_cast(&data), + static_cast(sizeof(int))); + + mqbi::StorageMessageAttributes attributes( + 0, + 2, // refCount: app1 and app2 + static_cast(appData->length()), + bmqp::MessagePropertiesInfo::makeNoSchema(), + bmqt::CompressionAlgorithmType::e_NONE); + + mqbi::DataStreamMessage* dataStreamMessage = 0; + BMQTST_ASSERT_EQ(storage.put(&attributes, + guid, + appData, + bsl::shared_ptr(), + &dataStreamMessage), + mqbi::StorageResult::e_SUCCESS); + BMQTST_ASSERT(dataStreamMessage); + + // Ordinals follow 'addVirtualStorage' order: app1 is 0, app2 is 1. + // app3's slot (ordinal 2) stays 'e_PUT'. + dataStreamMessage->app(0).setPushState(); + dataStreamMessage->app(1).setPushState(); + + // app3 never received the PUSH; its last reader goes away. + BMQTST_ASSERT(storage.removeVirtualStorage(k_APP_KEY3, false)); + + // app2 still holds the message, so app1's CONFIRM must not reach zero. + BMQTST_ASSERT_EQ(storage.confirm(guid, k_APP_KEY1, 0), + mqbi::StorageResult::e_NON_ZERO_REFERENCES); + BMQTST_ASSERT_EQ(storage.confirm(guid, k_APP_KEY2, 0), + mqbi::StorageResult::e_ZERO_REFERENCES); +} + BMQTST_TEST_F(Test, addQueueOpRecordHandle) { // CONSTANTS From 4576488d5671b0f4da612a0f1bd7285258a72aa2 Mon Sep 17 00:00:00 2001 From: Christopher Beard Date: Tue, 13 Oct 2026 08:44:05 -0400 Subject: [PATCH 2/5] IT[test_proxy_fanout_refcount]: expose ref loss for unpushed Apps On a proxy, a message's refCount includes only the Apps named in its PUSH. However, every App slot starts in e_PUT, which appears pending. When an App loses its last reader, the proxy removes its virtual storage. This wrongly releases references for messages that were never pushed to that App. The refCount then falls below the number of Apps that still hold the message. A subsequent CONFIRM can consequently delete the message while another App still needs it. If that App's copy is in the PushStream, the proxy aborts when it tries to load the deleted message. If the copy is in a redelivery list, the message is silently dropped. Add integration coverage for three variants: - An App closes while another App has a PushStream backlog. - An App crashes while another App has both PushStream and redelivery backlogs. - An App crashes while another App has only a redelivery backlog. The tests verify that the proxy remains alive and that every remaining App receives all its messages. Note that three apps are required to trigger the bug, the proxy does not assert for cases with only two Apps Signed-off-by: Christopher Beard --- .../test_proxy_fanout_refcount.py | 289 ++++++++++++++++++ 1 file changed, 289 insertions(+) create mode 100644 src/integration-tests/test_proxy_fanout_refcount.py diff --git a/src/integration-tests/test_proxy_fanout_refcount.py b/src/integration-tests/test_proxy_fanout_refcount.py new file mode 100644 index 0000000000..7b474b7019 --- /dev/null +++ b/src/integration-tests/test_proxy_fanout_refcount.py @@ -0,0 +1,289 @@ +# Copyright 2026 Bloomberg Finance L.P. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +""" +Fanout queues on a proxy: when an App's last reader goes away, the proxy must +keep the messages other Apps still need. + +A proxy keeps a fanout message until every App it was pushed to has +confirmed it. In each test, baz's last reader goes away while the proxy +holds messages that were not pushed to baz, then the test checks that foo +and bar still get every message. If the proxy treats baz as done with those +messages, foo's CONFIRMs delete messages bar still needs: the proxy either +aborts, when bar's copy is in the PushStream, or silently drops it, when +bar's copy is in a redelivery list. +""" + +import time + +import blazingmq.dev.it.testconstants as tc +from blazingmq.dev.it.fixtures import ( + Cluster, + order, + tweak, +) +from blazingmq.dev.configurator.configurator import Configurator +from blazingmq.dev.it.util import wait_until + +pytestmark = order(5) + +THROTTLE_INTERVAL_MS = 8000 + + +def long_message_throttling(configurator: Configurator): + """Delay every redelivery of a rejected message by the throttle interval.""" + throttle_config = configurator.proto.cluster.message_throttle_config + assert throttle_config is not None + throttle_config.high_threshold = 5 + throttle_config.low_threshold = 5 + throttle_config.low_interval = THROTTLE_INTERVAL_MS + throttle_config.high_interval = THROTTLE_INTERVAL_MS + + +def _payloads(client, uri): + return [m.payload for m in client.list(uri, block=True)] + + +def _wait_count(client, uri, count, timeout=5): + return wait_until(lambda: len(client.list(uri, block=True)) == count, timeout) + + +def _open_two_consumers_per_app(proxy, uri_foo, uri_bar): + consumers = {} + for name, app_uri in ( + ("foo1", uri_foo), + ("foo2", uri_foo), + ("bar1", uri_bar), + ("bar2", uri_bar), + ): + consumer = proxy.create_client(name) + consumer.open(app_uri, flags=["read"], max_unconfirmed_messages=3, succeed=True) + consumers[name] = (consumer, app_uri) + return consumers + + +def _consume_first_batch(producer, uri, baz, uri_baz, consumers): + """ + Post M1..M12. baz receives and confirms all of them, while each foo and + bar consumer takes 3 and confirms them. The proxy deletes M7..M12 once + baz confirms them, so they are pushed again to foo and bar only. + """ + for i in range(1, 13): + producer.post(uri, payload=[f"M{i}"], succeed=True) + + assert _wait_count(baz, uri_baz, 12) + baz.confirm(uri_baz, "*", succeed=True) + for consumer, app_uri in consumers.values(): + assert _wait_count(consumer, app_uri, 3) + consumer.confirm(app_uri, "*", succeed=True) + for consumer, app_uri in consumers.values(): + assert _wait_count(consumer, app_uri, 3) + + +@tweak.domain.max_delivery_attempts(5) +@tweak(long_message_throttling) +def test_reader_close_with_pushstream_backlog( + multi_node: Cluster, domain_urls: tc.DomainUrls +): + """ + Minimal sequence; bar's copies are in the PushStream. + + 1. foo (max 3), bar1 (max 2), bar2 (max 1), baz (max 1). Post + M1..M20: M1 is pushed to foo, bar and baz; M2 and M3 to foo and bar. + baz is full, so no later message is pushed to it. + 2. Kill bar1. Its two messages are rejected and wait, throttled, in + bar's redelivery list. + 3. Raise bar2's limit. The primary pushes M4..M20 to bar alone; bar's + redelivery list is not empty, so they stay in bar's PushStream. + 4. Raise foo's limit. M4..M20 are pushed to foo too. + 5. baz's consumer, its only reader, closes. + 6. foo confirms everything. + 7. When the throttle expires, bar redelivers and catches up: the proxy + must stay up and bar2 must get every message. + """ + du = domain_urls + uri = du.uri_fanout + uri_foo, uri_bar, uri_baz = du.uri_fanout_foo, du.uri_fanout_bar, du.uri_fanout_baz + proxy = next(multi_node.proxy_cycle()) + num_messages = 20 + + foo = proxy.create_client("foo") + foo.open(uri_foo, flags=["read"], max_unconfirmed_messages=3, succeed=True) + bar1 = proxy.create_client("bar1") + bar1.open(uri_bar, flags=["read"], max_unconfirmed_messages=2, succeed=True) + bar2 = proxy.create_client("bar2") + bar2.open(uri_bar, flags=["read"], max_unconfirmed_messages=1, succeed=True) + baz = proxy.create_client("baz") + baz.open(uri_baz, flags=["read"], max_unconfirmed_messages=1, succeed=True) + + producer = proxy.create_client("producer") + producer.open(uri, flags=["write", "ack"], succeed=True) + payloads = [f"M{i}" for i in range(1, num_messages + 1)] + for payload in payloads: + producer.post(uri, payload=[payload], succeed=True) + + assert _wait_count(foo, uri_foo, 3) + assert _wait_count(bar1, uri_bar, 2) + assert _wait_count(bar2, uri_bar, 1) + assert _wait_count(baz, uri_baz, 1) + assert sorted(_payloads(bar1, uri_bar) + _payloads(bar2, uri_bar)) == [ + "M1", + "M2", + "M3", + ] + assert _payloads(baz, uri_baz) == ["M1"] + + bar1.check_exit_code = False + bar1.kill() + assert proxy.outputs_substr("rejecting 2 messages because of a client crash", 10) + + bar2.configure(uri_bar, max_unconfirmed_messages=num_messages, succeed=True) + time.sleep(0.5) + foo.configure(uri_foo, max_unconfirmed_messages=num_messages, succeed=True) + assert _wait_count(foo, uri_foo, num_messages, 3) + # bar2 has its first message and one redelivery; M4..M20 wait in the + # PushStream. + assert _wait_count(bar2, uri_bar, 2) + + baz.close(uri_baz, succeed=True) + foo.confirm(uri_foo, "*", succeed=True) + + time.sleep(THROTTLE_INTERVAL_MS / 1000 + 2) + + assert proxy.is_alive() + assert wait_until(lambda: sorted(_payloads(bar2, uri_bar)) == sorted(payloads), 5) + + +@tweak.domain.max_delivery_attempts(5) +@tweak(long_message_throttling) +def test_reader_crash_with_pushstream_backlog( + multi_node: Cluster, domain_urls: tc.DomainUrls +): + """ + foo and bar are slow, baz is fast; baz's client crashes while bar has + copies both in a redelivery list and in the PushStream. + + 1. foo and bar have two consumers each (max 3). baz confirms M1..M12 + before M7..M12 are pushed again to foo and bar only (see + `_consume_first_batch`). + 2. Pause baz. Close foo2 and kill bar2, leaving a redelivery list of 3 + for each App. bar2's messages are rejected, so bar's redeliveries + are throttled. + 3. Raise foo1 and bar1's limits, then post N1..N6. Each N is pushed to + foo and bar together. foo takes it; bar, whose redelivery list is + not empty, falls behind in the PushStream. + 4. baz's client crashes holding no messages. + 5. foo confirms everything. + 6. bar redelivers and catches up: the proxy must stay up and bar1 must + get every message. + """ + du = domain_urls + uri = du.uri_fanout + uri_foo, uri_bar, uri_baz = du.uri_fanout_foo, du.uri_fanout_bar, du.uri_fanout_baz + proxy = next(multi_node.proxy_cycle()) + + consumers = _open_two_consumers_per_app(proxy, uri_foo, uri_bar) + foo1, foo2 = consumers["foo1"][0], consumers["foo2"][0] + bar1, bar2 = consumers["bar1"][0], consumers["bar2"][0] + baz = proxy.create_client("baz") + baz.open(uri_baz, flags=["read"], succeed=True) + producer = proxy.create_client("producer") + producer.open(uri, flags=["write", "ack"], succeed=True) + + _consume_first_batch(producer, uri, baz, uri_baz, consumers) + second_batch = sorted(f"M{i}" for i in range(7, 13)) + assert sorted(_payloads(foo1, uri_foo) + _payloads(foo2, uri_foo)) == second_batch + assert sorted(_payloads(bar1, uri_bar) + _payloads(bar2, uri_bar)) == second_batch + + baz.configure(uri_baz, max_unconfirmed_messages=0, succeed=True) + + foo2.close(uri_foo, succeed=True) + bar2.check_exit_code = False + bar2.kill() + assert proxy.outputs_substr("rejecting 3 messages because of a client crash", 10) + + bar1.configure(uri_bar, max_unconfirmed_messages=12, succeed=True) + foo1.configure(uri_foo, max_unconfirmed_messages=12, succeed=True) + assert _wait_count(foo1, uri_foo, 6, 3) + + for i in range(1, 7): + producer.post(uri, payload=[f"N{i}"], succeed=True) + assert _wait_count(foo1, uri_foo, 12, 3) + assert _wait_count(bar1, uri_bar, 4) + + baz.check_exit_code = False + baz.kill() + assert proxy.outputs_substr("rejecting 0 messages because of a client crash", 10) + + foo1.confirm(uri_foo, "*", succeed=True) + + time.sleep(THROTTLE_INTERVAL_MS / 1000 + 2) + + assert proxy.is_alive() + # bar2's other two messages are redelivered one throttle interval apart + # before bar catches up. + expected = sorted(second_batch + [f"N{i}" for i in range(1, 7)]) + assert wait_until( + lambda: sorted(_payloads(bar1, uri_bar)) == expected, + THROTTLE_INTERVAL_MS / 1000 + 5, + ) + + +def test_reader_crash_with_redelivery_backlog( + multi_node: Cluster, domain_urls: tc.DomainUrls +): + """ + bar's copies are only in a redelivery list, so the bug loses them + instead of aborting. + + 1. foo and bar have two consumers each (max 3). baz confirms M1..M12 + before M7..M12 are pushed again to foo and bar only (see + `_consume_first_batch`). + 2. Close foo2 and bar2, leaving a redelivery list of 3 for each App. + 3. baz's client crashes holding no messages. + 4. foo1 confirms its messages and those redelivered from foo2. + 5. bar1 confirms its messages: it must then get the 3 that bar2 held. + """ + du = domain_urls + uri = du.uri_fanout + uri_foo, uri_bar, uri_baz = du.uri_fanout_foo, du.uri_fanout_bar, du.uri_fanout_baz + proxy = next(multi_node.proxy_cycle()) + + consumers = _open_two_consumers_per_app(proxy, uri_foo, uri_bar) + foo1, foo2 = consumers["foo1"][0], consumers["foo2"][0] + bar1, bar2 = consumers["bar1"][0], consumers["bar2"][0] + baz = proxy.create_client("baz") + baz.open(uri_baz, flags=["read"], succeed=True) + producer = proxy.create_client("producer") + producer.open(uri, flags=["write", "ack"], succeed=True) + + _consume_first_batch(producer, uri, baz, uri_baz, consumers) + foo2_payloads = sorted(_payloads(foo2, uri_foo)) + bar2_payloads = sorted(_payloads(bar2, uri_bar)) + + foo2.close(uri_foo, succeed=True) + bar2.close(uri_bar, succeed=True) + + baz.check_exit_code = False + baz.kill() + assert proxy.outputs_substr("rejecting 0 messages because of a client crash", 10) + + foo1.confirm(uri_foo, "*", succeed=True) + assert wait_until(lambda: sorted(_payloads(foo1, uri_foo)) == foo2_payloads, 5) + foo1.confirm(uri_foo, "*", succeed=True) + + bar1.confirm(uri_bar, "*", succeed=True) + assert wait_until(lambda: sorted(_payloads(bar1, uri_bar)) == bar2_payloads, 5) + assert proxy.is_alive() From 13bb993d1e8169390b8b8ddba8078d28a7193520 Mon Sep 17 00:00:00 2001 From: Christopher Beard Date: Wed, 7 Oct 2026 18:05:01 -0400 Subject: [PATCH 3/5] Fix[mqbs]: proxy VS removal only releases refs of pushed Apps On a proxy, a message's refCount equals the number of Apps that received that message in a PUSH. The proxy marks exactly those Apps as 'e_PUSH'. Every other App slot keeps the initial 'e_PUT', which counts as pending. When an App lost its last reader, 'purgeImpl' released a reference for every pending slot, so messages never pushed to that App lost a reference. A later CONFIRM from one App then removed the message while another App still had it queued, leading to silent message loss or an assert in 'PushStreamIterator'. Add 'VirtualStorageCatalog::holdsReference' and release references only for App slots that hold one. Signed-off-by: Christopher Beard --- src/groups/mqb/mqbs/mqbs_virtualstoragecatalog.cpp | 9 +++++++-- src/groups/mqb/mqbs/mqbs_virtualstoragecatalog.h | 12 ++++++++++++ 2 files changed, 19 insertions(+), 2 deletions(-) diff --git a/src/groups/mqb/mqbs/mqbs_virtualstoragecatalog.cpp b/src/groups/mqb/mqbs/mqbs_virtualstoragecatalog.cpp index fad99cfe65..b5cdd63645 100644 --- a/src/groups/mqb/mqbs/mqbs_virtualstoragecatalog.cpp +++ b/src/groups/mqb/mqbs/mqbs_virtualstoragecatalog.cpp @@ -472,8 +472,13 @@ VirtualStorageCatalog::purgeImpl(VirtualStorage* vs, // Must call 'seek' before 'purge' setup(dataStreamMessage); - if (vs->remove(dataStreamMessage, replacingOrdinal)) { - // The 'data' was not already removed or confirmed. + // 'remove' resets the state, so check the reference first. + const bool holdsRef = holdsReference( + appMessageView(*dataStreamMessage, vs->ordinal())); + + if (vs->remove(dataStreamMessage, replacingOrdinal) && holdsRef) { + // The 'data' was not already removed or confirmed, and this App + // is counted in its refCount. result = d_storage_p->releaseRef(msgGUID, asPrimary); } diff --git a/src/groups/mqb/mqbs/mqbs_virtualstoragecatalog.h b/src/groups/mqb/mqbs/mqbs_virtualstoragecatalog.h index 37d20ffd7b..6a0834be82 100644 --- a/src/groups/mqb/mqbs/mqbs_virtualstoragecatalog.h +++ b/src/groups/mqb/mqbs/mqbs_virtualstoragecatalog.h @@ -390,6 +390,12 @@ class VirtualStorageCatalog BSLS_KEYWORD_FINAL { const mqbi::AppMessage& appMessageView(const mqbi::DataStreamMessage& dataStreamMessage, unsigned int ordinal) const; + + /// Return `true` if the specified `appMessage` accounts for a reference + /// in the refCount of its message, and `false` otherwise. A proxy + /// counts only the Apps a message was pushed to, while other nodes count + /// every App with a pending state. + bool holdsReference(const mqbi::AppMessage& appMessage) const; }; // ============================================================================ @@ -469,6 +475,12 @@ inline mqbi::Queue* VirtualStorageCatalog::queue() const return d_isProxy ? 0 : d_queue_p; } +inline bool +VirtualStorageCatalog::holdsReference(const mqbi::AppMessage& appMessage) const +{ + return d_isProxy ? appMessage.isPushing() : appMessage.isPending(); +} + } // close package namespace } // close enterprise namespace From 83bd52fd53cba4a4fbbc11f2cc7b2b68a243dc91 Mon Sep 17 00:00:00 2001 From: Christopher Beard Date: Fri, 16 Oct 2026 16:32:05 -0400 Subject: [PATCH 4/5] Refactor[mqbs]: call 'remove' for every message in purgeImpl 'remove' must run for every message, even when this App holds no reference, so explicitly call it on its own instead of inside of an if statement. After looking more closely though, the check can be simplified to just use 'holdsRef'; checking the return value of 'remove' is redundant. Signed-off-by: Christopher Beard --- src/groups/mqb/mqbs/mqbs_virtualstoragecatalog.cpp | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/src/groups/mqb/mqbs/mqbs_virtualstoragecatalog.cpp b/src/groups/mqb/mqbs/mqbs_virtualstoragecatalog.cpp index b5cdd63645..bb8c79122c 100644 --- a/src/groups/mqb/mqbs/mqbs_virtualstoragecatalog.cpp +++ b/src/groups/mqb/mqbs/mqbs_virtualstoragecatalog.cpp @@ -472,13 +472,15 @@ VirtualStorageCatalog::purgeImpl(VirtualStorage* vs, // Must call 'seek' before 'purge' setup(dataStreamMessage); - // 'remove' resets the state, so check the reference first. + // First, check whether this App is counted in the message's refCount. + // Then, call 'remove' for every message (even if this App does not + // hold a reference), which overwrites this App's state. const bool holdsRef = holdsReference( appMessageView(*dataStreamMessage, vs->ordinal())); - if (vs->remove(dataStreamMessage, replacingOrdinal) && holdsRef) { - // The 'data' was not already removed or confirmed, and this App - // is counted in its refCount. + vs->remove(dataStreamMessage, replacingOrdinal); + + if (holdsRef) { result = d_storage_p->releaseRef(msgGUID, asPrimary); } From 4963ebc4cac4e94f4f1c7645c7658d54e78e50fa Mon Sep 17 00:00:00 2001 From: Christopher Beard Date: Sat, 17 Oct 2026 15:37:34 -0400 Subject: [PATCH 5/5] IT[test_proxy_fanout_refcount]: CONFIRM after proxy restart must keep other Apps' messages After a proxy restarts, a consumer can confirm messages it was pushed before the restart. If the restarted proxy was pushed those messages for other Apps only, the CONFIRM from the unpushed App still releases a reference, so a later CONFIRM from one of those Apps deletes a message another App still has queued. This can cause a consumer to incorrectly never receive the the message. test_confirm_after_proxy_restart reproduces this bug using three Apps. This test fails until the fix in a following commit. Signed-off-by: Christopher Beard --- .../test_proxy_fanout_refcount.py | 77 +++++++++++++++++-- 1 file changed, 69 insertions(+), 8 deletions(-) diff --git a/src/integration-tests/test_proxy_fanout_refcount.py b/src/integration-tests/test_proxy_fanout_refcount.py index 7b474b7019..216c2c8c13 100644 --- a/src/integration-tests/test_proxy_fanout_refcount.py +++ b/src/integration-tests/test_proxy_fanout_refcount.py @@ -14,16 +14,17 @@ # limitations under the License. """ -Fanout queues on a proxy: when an App's last reader goes away, the proxy must -keep the messages other Apps still need. +Fanout queues on a proxy: the proxy must keep the messages other Apps still +need when an App's last reader goes away, or when an App confirms a message +that was not pushed to it. A proxy keeps a fanout message until every App it was pushed to has -confirmed it. In each test, baz's last reader goes away while the proxy -holds messages that were not pushed to baz, then the test checks that foo -and bar still get every message. If the proxy treats baz as done with those -messages, foo's CONFIRMs delete messages bar still needs: the proxy either -aborts, when bar's copy is in the PushStream, or silently drops it, when -bar's copy is in a redelivery list. +confirmed it. In each test, the proxy holds messages that were not pushed +to baz, and either baz's last reader goes away or baz confirms them. The +test then checks that foo and bar still get every message. If the proxy +treats baz as done with those messages, foo's CONFIRMs delete messages bar +still needs: the proxy either aborts, when bar's copy is in the PushStream, +or silently drops it, when bar's copy is in a redelivery list. """ import time @@ -287,3 +288,63 @@ def test_reader_crash_with_redelivery_backlog( bar1.confirm(uri_bar, "*", succeed=True) assert wait_until(lambda: sorted(_payloads(bar1, uri_bar)) == bar2_payloads, 5) assert proxy.is_alive() + + +def test_confirm_after_proxy_restart(multi_node: Cluster, domain_urls: tc.DomainUrls): + """ + baz confirms messages that the restarted proxy never pushed to it; bar's + copy is in a redelivery list. + + 1. Only baz consumes. It receives M1 and M2, then pauses (max 0). + 2. Restart the proxy. baz reconnects, still paused and still holding M1 + and M2. + 3. foo (max 2), bar1 and bar2 (max 1 each) open. The primary pushes M1 + and M2 to foo and bar only. + 4. Close bar2: its message waits in bar's redelivery list, as bar1 is + full. + 5. baz confirms M1 and M2. + 6. foo confirms both. bar1 confirms its message: it must then get the + one bar2 held. + """ + du = domain_urls + uri = du.uri_fanout + uri_foo, uri_bar, uri_baz = du.uri_fanout_foo, du.uri_fanout_bar, du.uri_fanout_baz + proxy = next(multi_node.proxy_cycle()) + + baz = proxy.create_client("baz") + baz.open(uri_baz, flags=["read"], succeed=True) + producer = proxy.create_client("producer") + producer.open(uri, flags=["write", "ack"], succeed=True) + producer.post(uri, payload=["M1"], succeed=True) + producer.post(uri, payload=["M2"], succeed=True) + assert _wait_count(baz, uri_baz, 2) + baz.configure(uri_baz, max_unconfirmed_messages=0, succeed=True) + + proxy.stop() + proxy.wait() + proxy.start() + proxy.wait_until_started() + proxy.wait_healthy() + baz.wait_state_restored() + producer.wait_state_restored() + + foo = proxy.create_client("foo") + foo.open(uri_foo, flags=["read"], max_unconfirmed_messages=2, succeed=True) + bar1 = proxy.create_client("bar1") + bar1.open(uri_bar, flags=["read"], max_unconfirmed_messages=1, succeed=True) + bar2 = proxy.create_client("bar2") + bar2.open(uri_bar, flags=["read"], max_unconfirmed_messages=1, succeed=True) + assert _wait_count(foo, uri_foo, 2) + assert _wait_count(bar1, uri_bar, 1) + assert _wait_count(bar2, uri_bar, 1) + bar2_payloads = _payloads(bar2, uri_bar) + + bar2.close(uri_bar, succeed=True) + + assert sorted(_payloads(baz, uri_baz)) == ["M1", "M2"] + baz.confirm(uri_baz, "*", succeed=True) + + foo.confirm(uri_foo, "*", succeed=True) + bar1.confirm(uri_bar, "*", succeed=True) + assert wait_until(lambda: _payloads(bar1, uri_bar) == bar2_payloads, 5) + assert proxy.is_alive()