gxs: batch message meta updates + bulk forum markRead (fix mark-all-as-read hang)

Marking a large forum (thousands of posts) "read" used to be catastrophic:
the GUI issued one blocking markRead() per post, and the GXS backend wrote
each status change to the SQLCipher DB in its own implicit transaction
(one fsync per message). On an 8000-post forum this took over an hour and,
if force-quit, left most posts still unread.

Backend changes:
- RsGeneralDataService: add a batch updateMessageMetaData(vector) overload.
  The default loops; RsDataService overrides it to persist the whole batch in
  a single transaction, held under one mDbMutex so nothing interleaves.
- RsGenExchange::processMsgMetaChanges(): resolve all status masks first, then
  persist every queued change through that single batched transaction.
- RsGxsForums / p3GxsForums: add a bulk markRead(forumId, msgIds, read) that
  queues all status changes at once and emits a single event, instead of one
  blocking request + one event per message.

"Mark all as read" now persists in a single transaction (one fsync) whatever
the number of posts.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: jolavillette

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/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/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;