Batch group meta data updates in a single DB transaction

processGrpMetaChanges() wrote each queued group meta update with its own
call to updateGroupMetaData(), i.e. one implicit SQLCipher transaction and
one fsync per entry. One such write was measured at ~1 s, and services
accumulating many updates (typically identity usage stamps at startup,
~3700 identity groups) froze their tick thread for as long as 86 s while
draining the queue, blocking all other GXS activity of the service through
mGenMtx.

Mirror the existing message-side batching: add a vector variant of
updateGroupMetaData() that wraps all row updates in a single transaction
(one fsync total), and make processGrpMetaChanges() collect the entries
that pass their mask and write them in one call.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

src/gxs/rsdataservice.cc

@@ -1583,6 +1583,56 @@ int RsDataService::updateGroupMetaData(const GrpLocMetaData& meta)
     return 0;
 }
 
+int RsDataService::updateGroupMetaData(const std::vector<GrpLocMetaData>& 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 group), which
+    // freezes the calling GXS service for seconds when many updates are queued
+    // (measured: ~1 s per update, 86 s backlogs on the identities service).
+    // We hold mDbMutex for the whole span so no other statement can slip into
+    // the transaction.
+    mDb->beginTransaction();
+
+    int count = 0;
+
+    for(const GrpLocMetaData& meta : metaList)
+    {
+        const RsGxsGroupId& grpId = meta.grpId;
+
+        if(mDb->sqlUpdate(GRP_TABLE_NAME, KEY_GRP_ID + "='" + grpId.toStdString() + "'", meta.val))
+        {
+            // If we use the cache, update the meta data immediately.
+            if(mUseCache)
+            {
+                RetroCursor* c = mDb->sqlQuery(GRP_TABLE_NAME, mGrpMetaColumns, "grpId='" + grpId.toStdString() + "'", "");
+
+                c->moveToFirst();
+
+                // temporarily disable the cache so that we get the value from the DB itself.
+                mUseCache=false;
+                auto meta_refreshed = locked_getGrpMeta(*c, 0);
+                mUseCache=true;
+
+                if(meta_refreshed)
+                    mGrpMetaDataCache.updateMeta(grpId,meta_refreshed);
+
+                delete c;
+            }
+
+            ++count;
+        }
+    }
+
+    mDb->commitTransaction();
+
+    return count;
+}
+
 int RsDataService::updateMessageMetaData(const MsgLocMetaData& metaData)
 {
 #ifdef RS_DATA_SERVICE_DEBUG_CACHE

src/gxs/rsdataservice.h

@@ -255,6 +255,8 @@ public:
      */
     int updateGroupMetaData(const GrpLocMetaData &meta) override;
 
+    int updateGroupMetaData(const std::vector<GrpLocMetaData>& metaList) override;
+
     /*!
      * Completely clear out data stored in
      * and returns this to a state

src/gxs/rsgds.h

@@ -266,6 +266,16 @@ public:
      */
     virtual int updateGroupMetaData(const GrpLocMetaData& meta) = 0;
 
+    /*!
+     * Update a batch of group meta data entries inside a single DB
+     * transaction. One transaction per entry means one fsync per entry,
+     * which freezes GXS services for seconds when many updates are queued
+     * (e.g. identity usage stamps at startup).
+     * @param metaList the meta data entries to update
+     * @return the number of entries successfully updated
+     */
+    virtual int updateGroupMetaData(const std::vector<GrpLocMetaData>& metaList) = 0;
+
     virtual int updateGroupKeys(const RsGxsGroupId& grpId,const RsTlvSecurityKeySet& keys,uint32_t subscribed_flags) = 0 ;
 
     /*!

src/gxs/rsgenexchange.cc

@@ -2205,6 +2205,13 @@ void RsGenExchange::processGrpMetaChanges()
 
     std::list<RsGxsGroupId> grpChanged;
 
+    // Phase 1: process the masks, and collect the entries to write.
+
+    std::vector<GrpLocMetaData> toWrite;
+    std::vector<uint32_t> writeTokens;
+    toWrite.reserve(metaMap.size());
+    writeTokens.reserve(metaMap.size());
+
     std::map<uint32_t, GrpLocMetaData>::iterator mit;
     for (mit = metaMap.begin(); mit != metaMap.end(); ++mit)
     {
@@ -2215,19 +2222,13 @@ void RsGenExchange::processGrpMetaChanges()
         RsDbg() << " Processing GrpMetaChange for token " << token << std::endl;
 #endif
         // process mask
-        bool ok = processGrpMask(g.grpId, g.val);
-
-        ok = ok && (mDataStore->updateGroupMetaData(g) == 1);
-
-        if(ok)
+        if(processGrpMask(g.grpId, g.val))
         {
-            mDataAccess->updatePublicRequestStatus(token, RsTokenService::COMPLETE);
-            grpChanged.push_back(g.grpId);
+            toWrite.push_back(g);
+            writeTokens.push_back(token);
         }
         else
-        {
             mDataAccess->updatePublicRequestStatus(token, RsTokenService::FAILED);
-        }
 
         {
             RS_STACK_MUTEX(mGenMtx);
@@ -2238,6 +2239,27 @@ void RsGenExchange::processGrpMetaChanges()
         }
     }
 
+    // Phase 2: write the whole batch in a single DB transaction. One call per
+    // entry means one fsync per entry, which was measured at ~1 s each and
+    // freezes the service tick for minutes when a backlog accumulates.
+
+    if(!toWrite.empty())
+    {
+        bool all_ok = mDataStore->updateGroupMetaData(toWrite) == (int)toWrite.size();
+
+        if(!all_ok)
+            RsErr() << __PRETTY_FUNCTION__ << " some group meta updates failed in a batch of " << toWrite.size() << " entries." << std::endl;
+
+        for(uint32_t i=0;i<toWrite.size();++i)
+            if(all_ok)
+            {
+                mDataAccess->updatePublicRequestStatus(writeTokens[i], RsTokenService::COMPLETE);
+                grpChanged.push_back(toWrite[i].grpId);
+            }
+            else
+                mDataAccess->updatePublicRequestStatus(writeTokens[i], RsTokenService::FAILED);
+    }
+
     for(auto& groupId:grpChanged)
     {
         RS_STACK_MUTEX(mGenMtx);