Fix: handle replication notifications on domain reconfigure

Signed-off-by: Evgeny Malygin <[email protected]>

src/groups/mqb/mqbblp/mqbblp_localqueue.cpp

@@ -185,6 +185,16 @@ int LocalQueue::configure(bsl::ostream& errorDescription, bool isReconfigure)
         domainCfg.storage().queueLimits().bytes());
 
     if (isReconfigure) {
+        if (domainCfg.consistency().isStrongValue()) {
+            /// We register notifications for weak consistency queues to
+            /// deliver messages after any replication.  For strong consistency
+            /// queues, we start to PUSH only on receipt.  We have to check and
+            /// possibly remove from notifications the queues that changed
+            /// their consistency level from weak to strong.
+            d_state_p->storageManager()->cancelReplicationNotification(
+                d_state_p->partitionId(),
+                d_state_p->key());
+        }
         if (domainCfg.mode().isFanoutValue()) {
             d_state_p->stats().updateDomainAppIds(
                 domainCfg.mode().fanout().appIDs());

src/groups/mqb/mqbblp/mqbblp_storagemanager.cpp

@@ -2130,6 +2130,21 @@ void StorageManager::gcUnrecognizedDomainQueues()
                                                   d_unrecognizedDomains);
 }
 
+void StorageManager::cancelReplicationNotification(
+    int                     partitionId,
+    const mqbu::StorageKey& queueKey)
+{
+    // executed by *QUEUE* dispatcher thread associated with partitionId
+
+    // PRECONDITIONS
+    BSLS_ASSERT_SAFE(0 <= partitionId &&
+                     partitionId < static_cast<int>(d_fileStores.size()));
+
+    mqbs::FileStore* fs = d_fileStores[partitionId].get();
+    BSLS_ASSERT_SAFE(fs->inDispatcherThread());
+    fs->cancelReplicationNotification(queueKey);
+}
+
 // ACCESSORS
 bool StorageManager::isStorageEmpty(const bmqt::Uri& uri,
                                     int              partitionId) const

src/groups/mqb/mqbblp/mqbblp_storagemanager.h

@@ -706,6 +706,13 @@ class StorageManager : public mqbi::StorageManager {
     /// GC the queues from unrecognized domains, if any.
     virtual void gcUnrecognizedDomainQueues() BSLS_KEYWORD_OVERRIDE;
 
+    /// Cancel replication notification (if any) for the specified `queueKey`.
+    /// Thread: executed by QUEUE dispatcher thread associated with the
+    ///         specified `partitionId`.
+    void cancelReplicationNotification(int                     partitionId,
+                                       const mqbu::StorageKey& queueKey)
+        BSLS_KEYWORD_OVERRIDE;
+
     // ACCESSORS
 
     /// Return the processor handle in charge of the specified

src/groups/mqb/mqbc/mqbc_storagemanager.cpp

@@ -4474,6 +4474,21 @@ void StorageManager::gcUnrecognizedDomainQueues()
                                             d_unrecognizedDomains);
 }
 
+void StorageManager::cancelReplicationNotification(
+    int                     partitionId,
+    const mqbu::StorageKey& queueKey)
+{
+    // executed by *QUEUE* dispatcher thread associated with partitionId
+
+    // PRECONDITIONS
+    BSLS_ASSERT_SAFE(0 <= partitionId &&
+                     partitionId < static_cast<int>(d_fileStores.size()));
+
+    mqbs::FileStore* fs = d_fileStores[partitionId].get();
+    BSLS_ASSERT_SAFE(fs->inDispatcherThread());
+    fs->cancelReplicationNotification(queueKey);
+}
+
 mqbs::FileStore& StorageManager::fileStore(int partitionId)
 {
     // PRECONDITIONS

src/groups/mqb/mqbc/mqbc_storagemanager.h

@@ -1046,6 +1046,13 @@ class StorageManager
     /// GC the queues from unrecognized domains, if any.
     virtual void gcUnrecognizedDomainQueues() BSLS_KEYWORD_OVERRIDE;
 
+    /// Cancel replication notification (if any) for the specified `queueKey`.
+    /// Thread: executed by QUEUE dispatcher thread associated with the
+    ///         specified `partitionId`.
+    void cancelReplicationNotification(int                     partitionId,
+                                       const mqbu::StorageKey& queueKey)
+        BSLS_KEYWORD_OVERRIDE;
+
     /// Return partition corresponding to the specified `partitionId`.  The
     /// behavior is undefined if `partitionId` does not represent a valid
     /// partition id. Note, this modifiable reference to partition is only

src/groups/mqb/mqbi/mqbi_storagemanager.h

@@ -413,6 +413,13 @@ class StorageManager {
     /// GC the queues from unrecognized domains, if any.
     virtual void gcUnrecognizedDomainQueues() = 0;
 
+    /// Cancel replication notification (if any) for the specified `queueKey`.
+    /// Thread: executed by QUEUE dispatcher thread associated with the
+    ///         specified `partitionId`.
+    virtual void
+    cancelReplicationNotification(int                     partitionId,
+                                  const mqbu::StorageKey& queueKey) = 0;
+
     // ACCESSORS
 
     /// Return the processor handle in charge of the specified

src/groups/mqb/mqbmock/mqbmock_storagemanager.cpp

@@ -276,6 +276,13 @@ void StorageManager::gcUnrecognizedDomainQueues()
     // NOTHING
 }
 
+void StorageManager::cancelReplicationNotification(
+    BSLS_ANNOTATION_UNUSED int   partitionId,
+    BSLS_ANNOTATION_UNUSED const mqbu::StorageKey& queueKey)
+{
+    // NOTHING
+}
+
 // ACCESSORS
 mqbi::Dispatcher::ProcessorHandle StorageManager::processorForPartition(
     BSLS_ANNOTATION_UNUSED int partitionId) const

src/groups/mqb/mqbmock/mqbmock_storagemanager.h

@@ -270,6 +270,13 @@ class StorageManager : public mqbi::StorageManager {
     /// GC the queues from unrecognized domains, if any.
     virtual void gcUnrecognizedDomainQueues() BSLS_KEYWORD_OVERRIDE;
 
+    /// Cancel replication notification (if any) for the specified `queueKey`.
+    /// Thread: executed by QUEUE dispatcher thread associated with the
+    ///         specified `partitionId`.
+    void cancelReplicationNotification(int                     partitionId,
+                                       const mqbu::StorageKey& queueKey)
+        BSLS_KEYWORD_OVERRIDE;
+
     // ACCESSORS
 
     /// Return the processor handle in charge of the specified

src/groups/mqb/mqbs/mqbs_filestore.cpp

@@ -6962,6 +6962,15 @@ void FileStore::notifyQueuesOnReplicatedBatch()
     }
 }
 
+void FileStore::cancelReplicationNotification(const mqbu::StorageKey& queueKey)
+{
+    bsl::unordered_set<mqbu::StorageKey>::iterator it =
+        d_replicationNotifications.find(queueKey);
+    if (it != d_replicationNotifications.end()) {
+        d_replicationNotifications.erase(it);
+    }
+}
+
 bool FileStore::gcExpiredMessages(const bdlt::Datetime& currentTimeUtc)
 {
     if (!d_isOpen) {

src/groups/mqb/mqbs/mqbs_filestore.h

@@ -882,6 +882,12 @@ class FileStore : public DataStore {
     /// builder is empty (just flushed).
     void notifyQueuesOnReplicatedBatch();
 
+    /// When reconfigure from weak consistency to strong consistency:
+    /// Cancel replication notification (if any) for the specified `queueKey`.
+    /// No effect if `queueKey` is not found in this storage or if there are no
+    /// current registered notifications for this queue.
+    void cancelReplicationNotification(const mqbu::StorageKey& queueKey);
+
     /// Invoke the specified `functor` with each queue associated to the
     /// partition represented by this FileStore if the partition was
     /// successfully opened.  The behavior is undefined unless invoked from