From 6a39b90a50768f8593318ceca87698d2c10cbe74 Mon Sep 17 00:00:00 2001 From: jolavillette Date: Sat, 11 Jul 2026 16:29:25 +0200 Subject: [PATCH 1/2] 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) --- src/gxs/rsdataservice.cc | 50 +++++++++++++++++++++++++++++++++ src/gxs/rsdataservice.h | 7 +++++ src/gxs/rsgds.h | 20 +++++++++++++ src/gxs/rsgenexchange.cc | 54 +++++++++++++++++++++++++----------- src/retroshare/rsgxsforums.h | 17 ++++++++++++ src/services/p3gxsforums.cc | 43 ++++++++++++++++++++++++++++ src/services/p3gxsforums.h | 5 ++++ 7 files changed, 180 insertions(+), 16 deletions(-) diff --git a/src/gxs/rsdataservice.cc b/src/gxs/rsdataservice.cc index 61900e171a..af1dd9ecd8 100644 --- a/src/gxs/rsdataservice.cc +++ b/src/gxs/rsdataservice.cc @@ -1581,6 +1581,56 @@ int RsDataService::updateMessageMetaData(const MsgLocMetaData& metaData) return 0; } +int RsDataService::updateMessageMetaData(const std::vector& 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); diff --git a/src/gxs/rsdataservice.h b/src/gxs/rsdataservice.h index e9554f04f8..3fd2fb7486 100644 --- a/src/gxs/rsdataservice.h +++ b/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& metaList) override; + /*! * @param metaData The meta data item to update * @return error code diff --git a/src/gxs/rsgds.h b/src/gxs/rsgds.h index 790a68deb8..eb88512da4 100644 --- a/src/gxs/rsgds.h +++ b/src/gxs/rsgds.h @@ -25,6 +25,7 @@ #include #include #include +#include #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& metaList) + { + int count = 0; + for(const MsgLocMetaData& m : metaList) + count += updateMessageMetaData(m); + return count; + } + /*! * @param metaData */ diff --git a/src/gxs/rsgenexchange.cc b/src/gxs/rsgenexchange.cc index c8773b91c0..5621e0448e 100644 --- a/src/gxs/rsgenexchange.cc +++ b/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 updates; + updates.reserve(metaMap.size()); + + std::vector > tokenIds; + tokenIds.reserve(metaMap.size()); + + std::set failedTokens; + std::map::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(updates.size())); + + for(std::vector >::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); diff --git a/src/retroshare/rsgxsforums.h b/src/retroshare/rsgxsforums.h index 3a9181916e..cbb0aab846 100644 --- a/src/retroshare/rsgxsforums.h +++ b/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& msgIds, + bool read ) = 0; + /** * @brief Subscrbe to a forum. Blocking API * @jsonapi{development} diff --git a/src/services/p3gxsforums.cc b/src/services/p3gxsforums.cc index a6c8aa73ff..bbe6a5ac49 100644 --- a/src/services/p3gxsforums.cc +++ b/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& 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 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(); + 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; diff --git a/src/services/p3gxsforums.h b/src/services/p3gxsforums.h index 31a458777d..5bd917b1a5 100644 --- a/src/services/p3gxsforums.h +++ b/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& msgIds, + bool read ) override; + /// @see RsGxsForums::updateReputationLevel virtual void updateReputationLevel(uint32_t forum_group_sign_flags,ForumPostEntry& e) const override; From 7047cd0a4c2747f7147c2ba740cb9fe78f445c88 Mon Sep 17 00:00:00 2001 From: jolavillette Date: Sun, 12 Jul 2026 00:03:14 +0200 Subject: [PATCH 2/2] gxs: bulk read-status for channels and boards (extend mark-all-as-read fix) Same treatment as the forum bulk markRead(), now for channels and boards, so their "mark all as read/unread" also persists in a single transaction and emits a single event instead of one blocking request + one event per post: - RsGxsChannels / p3GxsChannels: add setMessageReadStatus(channelId, msgIds, read). - RsPosted / p3Posted: add setPostReadStatus(boardId, msgIds, read). Both reuse the batched, single-transaction processMsgMetaChanges() path added for forums, which already benefits every GXS service. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/retroshare/rsgxschannels.h | 16 ++++++++++++++ src/retroshare/rsposted.h | 16 ++++++++++++++ src/services/p3gxschannels.cc | 40 ++++++++++++++++++++++++++++++++++ src/services/p3gxschannels.h | 5 +++++ src/services/p3posted.cc | 39 +++++++++++++++++++++++++++++++++ src/services/p3posted.h | 3 +++ 6 files changed, 119 insertions(+) diff --git a/src/retroshare/rsgxschannels.h b/src/retroshare/rsgxschannels.h index 4358e95c5a..c9df60830f 100644 --- a/src/retroshare/rsgxschannels.h +++ b/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& msgIds, + bool read ) =0; + /** * @brief Share channel publishing key * This can be used to authorize other peers to post on the channel diff --git a/src/retroshare/rsposted.h b/src/retroshare/rsposted.h index 412dc6f449..f304d7c41a 100644 --- a/src/retroshare/rsposted.h +++ b/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& msgIds, + bool read ) = 0; + enum RS_DEPRECATED RankType {TopRankType, HotRankType, NewRankType }; RS_DEPRECATED_FOR(getBoardsInfo) diff --git a/src/services/p3gxschannels.cc b/src/services/p3gxschannels.cc index e71fdb4775..b22e469209 100644 --- a/src/services/p3gxschannels.cc +++ b/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& 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 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(); + ev->mChannelGroupId = channelId; + ev->mChannelEventCode = RsChannelEventCode::READ_STATUS_CHANGED; + rsEvents->postEvent(ev); + } + + return true; +} + bool p3GxsChannels::shareChannelKeys( const RsGxsGroupId& channelId, const std::set& peers) { diff --git a/src/services/p3gxschannels.h b/src/services/p3gxschannels.h index ca15b0755c..99b5f6a510 100644 --- a/src/services/p3gxschannels.h +++ b/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& msgIds, + bool read ) override; + virtual bool groupShareKeys(const RsGxsGroupId &groupId, const std::set& peers) override; // no tokens... should be cached. diff --git a/src/services/p3posted.cc b/src/services/p3posted.cc index 2bb4f08471..442fa825d4 100644 --- a/src/services/p3posted.cc +++ b/src/services/p3posted.cc @@ -707,6 +707,45 @@ bool p3Posted::setPostReadStatus(const RsGxsGrpMsgIdPair &msgId, bool read) { return setCommentReadStatus(msgId,read); } +bool p3Posted::setPostReadStatus(const RsGxsGroupId& boardId, const std::vector& 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 tokens; + tokens.reserve(msgIds.size()); + + for(const RsGxsMessageId& msgId : msgIds) + { + uint32_t token; + setMsgStatusFlags(token, RsGxsGrpMsgIdPair(boardId, 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(); + ev->mPostedGroupId = boardId; + ev->mPostedEventCode = RsPostedEventCode::READ_STATUS_CHANGED; + rsEvents->postEvent(ev); + } + + return true; +} bool p3Posted::setCommentReadStatus(const RsGxsGrpMsgIdPair &msgId, bool read) { uint32_t token; diff --git a/src/services/p3posted.h b/src/services/p3posted.h index 4adc2cf299..0656f1fe10 100644 --- a/src/services/p3posted.h +++ b/src/services/p3posted.h @@ -125,6 +125,9 @@ virtual void receiveHelperChanges(std::vector& changes) bool setCommentReadStatus(const RsGxsGrpMsgIdPair &msgId, bool read) override; bool setPostReadStatus(const RsGxsGrpMsgIdPair &msgId, bool read) override; + bool setPostReadStatus( const RsGxsGroupId& boardId, + const std::vector& msgIds, + bool read ) override; bool getRelatedComments( const RsGxsGroupId& gid,const std::set& msgIds, std::vector &comments ) override;