Fix UI freeze on "mark all as read/unread" for forums, channels and boards

src/gxs/rsdataservice.cc

@@ -1581,6 +1581,56 @@ int RsDataService::updateMessageMetaData(const MsgLocMetaData& metaData)
     return 0;
 }
 
+int RsDataService::updateMessageMetaData(const std::vector<MsgLocMetaData>& metaList)
+{
+    if(metaList.empty())
+        return 0;
+
+    RsStackMutex stack(mDbMutex);
+
+    // Persist the whole batch inside a single transaction. Without this, every
+    // row update is its own implicit transaction (one fsync per message), which
+    // is what made "mark all as read" take more than an hour on a large forum.
+    // We hold mDbMutex for the whole span so no other statement can slip into
+    // the transaction.
+    mDb->beginTransaction();
+
+    int count = 0;
+
+    for(const MsgLocMetaData& metaData : metaList)
+    {
+        const RsGxsGroupId& grpId = metaData.msgId.first;
+        const RsGxsMessageId& msgId = metaData.msgId.second;
+
+        if(mDb->sqlUpdate(MSG_TABLE_NAME,  KEY_GRP_ID+ "='" + grpId.toStdString() + "' AND " + KEY_MSG_ID + "='" + msgId.toStdString() + "'", metaData.val) )
+        {
+            // If we use the cache, update the meta data immediately.
+            if(mUseCache)
+            {
+                RetroCursor* c = mDb->sqlQuery(MSG_TABLE_NAME, mMsgMetaColumns, KEY_GRP_ID+ "='" + grpId.toStdString() + "' AND " + KEY_MSG_ID + "='" + msgId.toStdString() + "'", "");
+
+                c->moveToFirst();
+
+                // temporarily disable the cache so that we get the value from the DB itself.
+                mUseCache=false;
+                auto meta = locked_getMsgMeta(*c, 0);
+                mUseCache=true;
+
+                if(meta)
+                    mMsgMetaDataCache[grpId].updateMeta(msgId,meta);
+
+                delete c;
+            }
+
+            ++count;
+        }
+    }
+
+    mDb->commitTransaction();
+
+    return count;
+}
+
 int RsDataService::removeMsgs(const GxsMsgReq& msgIds)
 {
     RsStackMutex stack(mDbMutex);

src/gxs/rsdataservice.h

@@ -242,6 +242,13 @@ class RsDataService : public RsGeneralDataService
      */
     int updateMessageMetaData(const MsgLocMetaData& metaData) override;
 
+    /*!
+     * @brief Batch variant persisting all updates in a single DB transaction.
+     * @param metaList The meta data items to update
+     * @return the number of items successfully updated
+     */
+    int updateMessageMetaData(const std::vector<MsgLocMetaData>& metaList) override;
+
     /*!
      * @param metaData The meta data item to update
      * @return error code

src/gxs/rsgds.h

@@ -25,6 +25,7 @@
 #include <set>
 #include <map>
 #include <string>
+#include <vector>
 
 #include "inttypes.h"
 
@@ -248,6 +249,25 @@ class RsGeneralDataService
      */
     virtual int updateMessageMetaData(const MsgLocMetaData& metaData) = 0;
 
+    /*!
+     * @brief Update the meta data of several messages at once.
+     *
+     * The default implementation just updates them one by one. Stores backed by
+     * a transactional database should override this to persist the whole batch
+     * in a single transaction, which is dramatically faster when marking
+     * thousands of messages (e.g. "mark all as read" on a large forum).
+     *
+     * @param metaList the meta data items to update
+     * @return the number of items successfully updated
+     */
+    virtual int updateMessageMetaData(const std::vector<MsgLocMetaData>& metaList)
+    {
+        int count = 0;
+        for(const MsgLocMetaData& m : metaList)
+            count += updateMessageMetaData(m);
+        return count;
+    }
+
     /*!
      * @param metaData
      */

src/gxs/rsgenexchange.cc

@@ -2093,6 +2093,18 @@ void RsGenExchange::processMsgMetaChanges()
 
     GxsMsgReq msgIds;
 
+    // First pass: resolve the status masks into absolute values (reads come from
+    // the in-memory meta cache, so this stays cheap even for thousands of
+    // messages) and collect every change so they can be persisted together in a
+    // single DB transaction below, instead of one fsync'd write per message.
+    std::vector<MsgLocMetaData> updates;
+    updates.reserve(metaMap.size());
+
+    std::vector<std::pair<uint32_t, RsGxsGrpMsgIdPair> > tokenIds;
+    tokenIds.reserve(metaMap.size());
+
+    std::set<uint32_t> failedTokens;
+
     std::map<uint32_t, MsgLocMetaData>::iterator mit;
     for (mit = metaMap.begin(); mit != metaMap.end(); ++mit)
     {
@@ -2133,28 +2145,38 @@ void RsGenExchange::processMsgMetaChanges()
             }
         }
 
-        ok &= mDataStore->updateMessageMetaData(m) == 1;
-        uint32_t token = mit->first;
+        // The actual DB write is deferred to the single batched transaction
+        // below. Collect the resolved change and remember its token.
+        updates.push_back(m);
+        tokenIds.push_back(std::make_pair(mit->first, m.msgId));
+
+        if(!ok)
+            failedTokens.insert(mit->first);
+        else if(changed)
+            msgIds[m.msgId.first].insert(m.msgId.second);
+    }
+
+    // Second pass: persist every collected change in a single transaction.
+    int updated = mDataStore->updateMessageMetaData(updates);
+    bool batchOk = (updated == static_cast<int>(updates.size()));
+
+    for(std::vector<std::pair<uint32_t, RsGxsGrpMsgIdPair> >::iterator it = tokenIds.begin(); it != tokenIds.end(); ++it)
+    {
+        bool ok = batchOk && (failedTokens.find(it->first) == failedTokens.end());
 
         if(ok)
-        {
-            mDataAccess->updatePublicRequestStatus(token, RsTokenService::COMPLETE);
-            if (changed)
-            {
-                msgIds[m.msgId.first].insert(m.msgId.second);
-            }
-        }
+            mDataAccess->updatePublicRequestStatus(it->first, RsTokenService::COMPLETE);
         else
-        {
-            mDataAccess->updatePublicRequestStatus(token, RsTokenService::FAILED);
-        }
+            mDataAccess->updatePublicRequestStatus(it->first, RsTokenService::FAILED);
 
-        {
-            RS_STACK_MUTEX(mGenMtx);
-            mMsgNotify.insert(std::make_pair(token, m.msgId));
-        }
+        RS_STACK_MUTEX(mGenMtx);
+        mMsgNotify.insert(std::make_pair(it->first, it->second));
     }
 
+    // If the whole batch failed, don't emit change notifications for it.
+    if(!batchOk)
+        msgIds.clear();
+
     if (!msgIds.empty())
     {
         RS_STACK_MUTEX(mGenMtx);

src/retroshare/rsgxschannels.h

@@ -430,6 +430,22 @@ class RsGxsChannels: public RsGxsIfaceHelper, public RsGxsCommentService
      */
     virtual bool setMessageReadStatus(const RsGxsGrpMsgIdPair &msgId, bool read) =0;
 
+    /**
+     * @brief Mark several messages of a channel read/unread in one batch. Blocking.
+     *
+     * Like setMessageReadStatus() but queues all the status changes at once: they
+     * are persisted in a single database transaction and only one event is
+     * emitted, so it stays cheap even for thousands of posts and is meant to be
+     * called from a background thread so the GUI never blocks.
+     * @param[in] channelId channel group identifier
+     * @param[in] msgIds     identifiers of the posts to update
+     * @param[in] read       true to mark as read, false to mark as unread
+     * @return false on error, true otherwise
+     */
+    virtual bool setMessageReadStatus( const RsGxsGroupId& channelId,
+                                       const std::vector<RsGxsMessageId>& msgIds,
+                                       bool read ) =0;
+
     /**
 	 * @brief Share channel publishing key
 	 * This can be used to authorize other peers to post on the channel

src/retroshare/rsgxsforums.h

@@ -407,6 +407,23 @@ class RsGxsForums: public RsGxsIfaceHelper
 	 */
 	virtual bool markRead(const RsGxsGrpMsgIdPair& messageId, bool read) = 0;
 
+	/**
+	 * @brief Mark several messages of a forum read/unread in one batch. Blocking.
+	 *
+	 * Contrary to calling markRead() in a loop, this queues all the status
+	 * changes at once: they are persisted in a single database transaction and
+	 * only one event is emitted. It therefore stays cheap even for thousands of
+	 * posts (e.g. "mark all as read" on a large forum) and is meant to be called
+	 * from a background thread so the caller (typically the GUI) never blocks.
+	 * @param[in] forumId forum group identifier
+	 * @param[in] msgIds  identifiers of the posts to update
+	 * @param[in] read    true to mark as read, false to mark as unread
+	 * @return false on error, true otherwise
+	 */
+	virtual bool markRead( const RsGxsGroupId& forumId,
+	                       const std::vector<RsGxsMessageId>& msgIds,
+	                       bool read ) = 0;
+
 	/**
 	 * @brief Subscrbe to a forum. Blocking API
 	 * @jsonapi{development}

src/retroshare/rsposted.h

@@ -402,6 +402,22 @@ class RsPosted : public RsGxsIfaceHelper, public RsGxsCommentService
      */
     virtual bool setPostReadStatus(const RsGxsGrpMsgIdPair& msgId, bool read) = 0;
 
+    /**
+     * @brief Mark several posts of a board read/unread in one batch. Blocking.
+     *
+     * Like setPostReadStatus() but queues all the status changes at once: they
+     * are persisted in a single database transaction and only one event is
+     * emitted, so it stays cheap even for thousands of posts and is meant to be
+     * called from a background thread so the GUI never blocks.
+     * @param[in] boardId board group identifier
+     * @param[in] msgIds  identifiers of the posts to update
+     * @param[in] read    true to mark as read, false to mark as unread
+     * @return false on error, true otherwise
+     */
+    virtual bool setPostReadStatus( const RsGxsGroupId& boardId,
+                                    const std::vector<RsGxsMessageId>& msgIds,
+                                    bool read ) = 0;
+
 	enum RS_DEPRECATED RankType {TopRankType, HotRankType, NewRankType };
 
 	RS_DEPRECATED_FOR(getBoardsInfo)

src/services/p3gxschannels.cc

@@ -2041,6 +2041,46 @@ bool p3GxsChannels::setMessageReadStatus(const RsGxsGrpMsgIdPair &msgId, bool re
 	return true;
 }
 
+bool p3GxsChannels::setMessageReadStatus(const RsGxsGroupId& channelId, const std::vector<RsGxsMessageId>& msgIds, bool read)
+{
+	if(msgIds.empty())
+		return true;
+
+	uint32_t mask   = GXS_SERV::GXS_MSG_STATUS_GUI_NEW | GXS_SERV::GXS_MSG_STATUS_GUI_UNREAD;
+	uint32_t status = read ? 0 : GXS_SERV::GXS_MSG_STATUS_GUI_UNREAD;
+
+	// Queue every status change at once. They are all drained together by a
+	// single processMsgMetaChanges() tick (one DB transaction), instead of one
+	// blocking request + one detached thread + one event per message.
+	std::vector<uint32_t> tokens;
+	tokens.reserve(msgIds.size());
+
+	for(const RsGxsMessageId& msgId : msgIds)
+	{
+		uint32_t token;
+		setMsgStatusFlags(token, RsGxsGrpMsgIdPair(channelId, msgId), status, mask);
+		tokens.push_back(token);
+	}
+
+	// All tokens complete in the same tick, so waiting on the last is enough.
+	waitToken(tokens.back(), std::chrono::milliseconds(30000));
+
+	RsGxsGrpMsgIdPair p;
+	for(uint32_t token : tokens)
+		acknowledgeMsg(token, p);
+
+	// One single event for the whole batch (consumers only need the group id).
+	if(rsEvents)
+	{
+		auto ev = std::make_shared<RsGxsChannelEvent>();
+		ev->mChannelGroupId   = channelId;
+		ev->mChannelEventCode = RsChannelEventCode::READ_STATUS_CHANGED;
+		rsEvents->postEvent(ev);
+	}
+
+	return true;
+}
+
 bool p3GxsChannels::shareChannelKeys(
         const RsGxsGroupId& channelId, const std::set<RsPeerId>& peers)
 {

src/services/p3gxschannels.h

@@ -106,6 +106,11 @@ class p3GxsChannels: public RsGenExchange, public RsGxsChannels,
     /// @see RsGxsChannels
     bool setMessageReadStatus(const RsGxsGrpMsgIdPair& msgId, bool read) override;
 
+    /// @see RsGxsChannels (batch variant)
+    bool setMessageReadStatus( const RsGxsGroupId& channelId,
+                               const std::vector<RsGxsMessageId>& msgIds,
+                               bool read ) override;
+
     virtual bool groupShareKeys(const RsGxsGroupId &groupId, const std::set<RsPeerId>& peers) override;
 
     // no tokens... should be cached.

src/services/p3gxsforums.cc

@@ -895,6 +895,49 @@ bool p3GxsForums::markRead(const RsGxsGrpMsgIdPair& msgId, bool read)
 	return true;
 }
 
+bool p3GxsForums::markRead(const RsGxsGroupId& forumId, const std::vector<RsGxsMessageId>& msgIds, bool read)
+{
+	if(msgIds.empty())
+		return true;
+
+	uint32_t mask   = GXS_SERV::GXS_MSG_STATUS_GUI_NEW | GXS_SERV::GXS_MSG_STATUS_GUI_UNREAD;
+	uint32_t status = read ? 0 : GXS_SERV::GXS_MSG_STATUS_GUI_UNREAD;
+
+	// Queue every status change at once. They all land in mMsgLocMetaMap and are
+	// drained together by a single processMsgMetaChanges() tick, which persists
+	// them in one DB transaction. This is what keeps "mark all as read" cheap
+	// instead of one blocking request (and one detached thread) per message.
+	std::vector<uint32_t> tokens;
+	tokens.reserve(msgIds.size());
+
+	for(const RsGxsMessageId& msgId : msgIds)
+	{
+		uint32_t token;
+		setMsgStatusFlags(token, RsGxsGrpMsgIdPair(forumId, msgId), status, mask);
+		tokens.push_back(token);
+	}
+
+	// Wait for the whole batch to be processed. All tokens complete in the same
+	// tick, so waiting on the last one is enough.
+	waitToken(tokens.back(), std::chrono::milliseconds(30000));
+
+	RsGxsGrpMsgIdPair p;
+	for(uint32_t token : tokens)
+		acknowledgeMsg(token, p);
+
+	// A single event for the whole batch (the per-message event storm was part
+	// of what froze the UI). Consumers only use the group id to refresh counts.
+	if(rsEvents)
+	{
+		auto ev = std::make_shared<RsGxsForumEvent>();
+		ev->mForumGroupId   = forumId;
+		ev->mForumEventCode = RsForumEventCode::READ_STATUS_CHANGED;
+		rsEvents->postEvent(ev);
+	}
+
+	return true;
+}
+
 bool p3GxsForums::subscribeToForum(const RsGxsGroupId& groupId, bool subscribe )
 {
 	uint32_t token;

src/services/p3gxsforums.h

@@ -124,6 +124,11 @@ class p3GxsForums: public RsGenExchange, public RsGxsForums, public p3Config,
 	/// @see RsGxsForums::markRead
     virtual bool markRead(const RsGxsGrpMsgIdPair& messageId, bool read) override;
 
+	/// @see RsGxsForums::markRead (batch variant)
+    virtual bool markRead( const RsGxsGroupId& forumId,
+                           const std::vector<RsGxsMessageId>& msgIds,
+                           bool read ) override;
+
     /// @see RsGxsForums::updateReputationLevel
     virtual void updateReputationLevel(uint32_t forum_group_sign_flags,ForumPostEntry& e) const override;