gc'ing old message in the presence of new app

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

Author: dorjesinpo

src/groups/mqb/mqbc/mqbc_storageutil.cpp

@@ -1521,7 +1521,7 @@ void StorageUtil::recoveredQueuesCb(
                         << "encountered a duplicate AppId while processing "
                         << "recovered queue [" << uri << "], " << "queueKey ["
                         << qit->first << "]. AppId [" << *(appIdsIrc.first)
-                        << "]. AppKey [" << cit->second << "]."
+                        << "]. AppKey [" << cit->first << "]."
                         << BMQTSK_ALARMLOG_END;
                     mqbu::ExitUtil::terminate(
                         mqbu::ExitCode::e_RECOVERY_FAILURE);

src/groups/mqb/mqbs/mqbs_virtualstorage.cpp

@@ -142,16 +142,10 @@ bool VirtualStorage::remove(mqbi::DataStreamMessage* dataStreamMessage,
     return wasPending;
 }
 
-void VirtualStorage::onGC(const mqbi::DataStreamMessage& dataStreamMessage)
+void VirtualStorage::onGC(int size)
 {
-    if (!dataStreamMessage.d_apps.empty()) {
-        const mqbi::AppMessage& appMessage = dataStreamMessage.app(ordinal());
-
-        if (!appMessage.isPending()) {
-            d_removedBytes -= dataStreamMessage.d_size;
-            --d_numRemoved;
-        }
-    }
+    d_removedBytes -= size;
+    --d_numRemoved;
 }
 
 void VirtualStorage::resetStats()
@@ -165,9 +159,11 @@ void VirtualStorage::replaceOrdinal(unsigned int replacingOrdinal)
     d_ordinal = replacingOrdinal;
 }
 
-void VirtualStorage::setNumRemoved(bsls::Types::Int64 numRemoved)
+void VirtualStorage::setNumRemoved(bsls::Types::Int64 numMessages,
+                                   bsls::Types::Int64 bytes)
 {
-    d_numRemoved = numRemoved;
+    d_numRemoved   = numMessages;
+    d_removedBytes = bytes;
 }
 
 bool VirtualStorage::hasReceipt(const bmqt::MessageGUID& msgGUID) const

src/groups/mqb/mqbs/mqbs_virtualstorage.h

@@ -172,16 +172,17 @@ class VirtualStorage {
     bool remove(mqbi::DataStreamMessage* dataStreamMessage,
                 unsigned int             replacingOrdinal);
 
-    /// Observe removal of this App from the specified 'dataStreamMessage' by
-    /// GC and update bytes and messages counts if needed.
-    void onGC(const mqbi::DataStreamMessage& dataStreamMessage);
+    /// Update bytes by the specified 'size' and messages counts by `1` for
+    /// this App as the result of garbage-collecting the message.
+    void onGC(int size);
 
     /// Reset bytes and messages counts as in the case of purging all Apps.
     void resetStats();
 
     void replaceOrdinal(unsigned int replacingOrdinal);
 
-    void setNumRemoved(bsls::Types::Int64 numRemoved);
+    void setNumRemoved(bsls::Types::Int64 numRemoved,
+                       bsls::Types::Int64 bytes);
 };
 
 // =====================

src/groups/mqb/mqbs/mqbs_virtualstoragecatalog.cpp

@@ -312,7 +312,13 @@ VirtualStorageCatalog::gc(const bmqt::MessageGUID& msgGUID)
     for (VirtualStoragesIter it = d_virtualStorages.begin();
          it != d_virtualStorages.end();
          ++it) {
-        it->value()->onGC(data->second);
+        const mqbi::DataStreamMessage& dataStreamMessage = data->second;
+
+        const mqbi::AppMessage& appMessage =
+            appMessageView(dataStreamMessage, it->value()->ordinal());
+        if (!appMessage.isPending()) {
+            it->value()->onGC(dataStreamMessage.d_size);
+        }
     }
 
     d_totalBytes -= data->second.d_size;
@@ -351,7 +357,8 @@ void VirtualStorageCatalog::removeAll(const mqbu::StorageKey& appKey)
 }
 
 bsls::Types::Int64 VirtualStorageCatalog::seek(DataStreamIterator*   it,
-                                               const VirtualStorage* vs)
+                                               const VirtualStorage* vs,
+                                               bsls::Types::Int64*   bytes)
 {
     BSLS_ASSERT_SAFE(vs);
 
@@ -372,6 +379,11 @@ bsls::Types::Int64 VirtualStorageCatalog::seek(DataStreamIterator*   it,
             break;  // BREAK
         }
         // else 'dataStreamMessage' is older than this VirtualStorage
+
+        if (BSLS_PERFORMANCEHINT_PREDICT_UNLIKELY(bytes)) {
+            BSLS_PERFORMANCEHINT_UNLIKELY_HINT;
+            *bytes += data.d_size;
+        }
     }
 
     return result;
@@ -663,7 +675,11 @@ void VirtualStorageCatalog::calibrate()
          it != d_virtualStorages.end();
          ++it) {
         DataStreamIterator itData;
-        it->value()->setNumRemoved(seek(&itData, it->value().get()));
+        bsls::Types::Int64 bytes       = 0;
+        bsls::Types::Int64 numMessages = seek(&itData,
+                                              it->value().get(),
+                                              &bytes);
+        it->value()->setNumRemoved(numMessages, bytes);
     }
 }
 

src/groups/mqb/mqbs/mqbs_virtualstoragecatalog.h

@@ -255,8 +255,12 @@ class VirtualStorageCatalog {
 
     /// Return the number of messages in the datastream which are older than
     /// the specified `vs`.  Load into the specified `it` the iterator pointing
-    /// either to the first newer message or to the end of the datastream.
-    bsls::Types::Int64 seek(DataStreamIterator* it, const VirtualStorage* vs);
+    /// either to the first newer message or to the end of the datastream.  If
+    /// the optionally specified `bytes` is not `0`, load the sum of relevant
+    /// messages.
+    bsls::Types::Int64 seek(DataStreamIterator*   it,
+                            const VirtualStorage* vs,
+                            bsls::Types::Int64*   bytes = 0);
 
     /// Create, if it doesn't exist already, a virtual storage instance with
     /// the specified 'appId' and 'appKey'.  Return zero upon success and a

src/integration-tests/test_appids.py

@@ -871,3 +871,55 @@ def _verify_clients(andConfirm=False):
 
     leader.list_messages(du.domain_fanout, tc.TEST_QUEUE, 0, 100)
     assert leader.outputs_substr(f"Printing 0 message(s)", 5)
+
+
+@tweak.domain.message_ttl(3)
+def test_gc_old_data_new_app(cluster: Cluster, domain_urls: tc.DomainUrls):
+    """Trigger old message GC in the presence of new App.  Need to allocate
+    Apps states first.
+    """
+    du = domain_urls
+    leader = cluster.last_known_leader
+    proxies = cluster.proxy_cycle()
+    proxy = next(proxies)
+
+    producer = proxy.create_client("producer")
+    producer.open(du.uri_fanout, flags=["write,ack"], succeed=True)
+
+    app_id = default_app_ids[0]
+    consumer = proxy.create_client(app_id)
+    consumer_uri = f"{du.uri_fanout}?id={app_id}"
+    consumer.open(consumer_uri, flags=["read"], succeed=True)
+
+    # ---------------------------------------------------------------------
+    # Post a message.
+    producer.post(du.uri_fanout, ["m1"], succeed=True, wait_ack=True)
+
+    # confirm, to make sure App state is allocated
+    consumer.wait_push_event()
+
+    assert wait_until(
+        lambda: len(consumer.list(consumer_uri, block=True)) == 1, timeout
+    )
+    assert consumer.confirm(consumer_uri, f"+{1}", block=True) == Client.e_SUCCESS
+
+    # ---------------------------------------------------------------------
+    # +new_app_1
+    new_app_1 = "new_app_1"
+    set_app_ids(cluster, default_app_ids + [new_app_1], du)
+
+    assert consumer.close(consumer_uri, block=True) == Client.e_SUCCESS
+
+    # Observe that the message was GC'd from the queue.
+    leader.capture(
+        f"queue \\[{du.uri_fanout}\\].*garbage-collected \\[1\\] messages", timeout=5
+    )
+
+    leader.list_messages(du.domain_fanout, tc.TEST_QUEUE, 0, 100)
+    assert leader.outputs_substr(f"Printing 0 message(s)", 5)
+
+    leader.list_messages(du.domain_fanout, tc.TEST_QUEUE, 0, 100, appid=app_id)
+    assert leader.outputs_substr(f"Printing 0 message(s)", 5)
+
+    leader.list_messages(du.domain_fanout, tc.TEST_QUEUE, 0, 100, appid=new_app_1)
+    assert leader.outputs_substr(f"Printing 0 message(s)", 5)