core: roster transfers (#7089)

src/Simplex/Chat/Library/Internal.hs

@@ -1884,14 +1884,30 @@ closeFileHandle fileId files = do
 
 -- The roster file has no chat item, so chat-item file enumeration misses it; clean it up by group.
 cleanupGroupRosterFile :: User -> GroupInfo -> CM ()
-cleanupGroupRosterFile User {userId} gInfo@GroupInfo {groupId} = do
+cleanupGroupRosterFile User {userId} GroupInfo {groupId} = do
   infos <- withStore' $ \db -> getGroupRosterFileInfo db userId groupId
   forM_ infos $ \(fileId, filePath_) -> do
     lift $ closeFileHandle fileId rcvFiles
     forM_ filePath_ removeFsFile
   withStore' $ \db -> do
     deleteGroupRosterFile db userId groupId
-    clearRosterPending db gInfo
+    deleteGroupRosterTransfers db groupId
+
+-- Supersede/cancel one source relay's in-flight roster transfer: remove its on-disk file + cached
+-- handle first (the cascade only does rows), then the files + transfer rows.
+cleanupRosterTransfer :: GroupInfo -> GroupMemberId -> CM ()
+cleanupRosterTransfer gInfo fromMemberId =
+  withStore' (\db -> getRosterTransferId db gInfo fromMemberId) >>= mapM_ cleanupRosterTransferById
+
+cleanupRosterTransferById :: Int64 -> CM ()
+cleanupRosterTransferById transferId = do
+  file_ <- withStore' $ \db -> getRosterTransferFile db transferId
+  forM_ file_ $ \(fileId, filePath_) -> do
+    lift $ closeFileHandle fileId rcvFiles
+    forM_ filePath_ removeFsFile
+  withStore' $ \db -> do
+    deleteRosterTransferFile db transferId
+    deleteRosterTransfer db transferId
 
 -- MUST evict the cached AppendMode handle before deleting chunks, else re-driven bytes append
 -- after the stale prefix and corrupt the blob.

src/Simplex/Chat/Library/Subscriber.hs

@@ -1084,7 +1084,7 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage =
               XGrpDel -> Just (DeliveryTaskContext (DJSGroup {jobSpec = DJRelayRemoved}) False) <$ xGrpDel gInfo' m'' msg brokerTs
               XGrpInfo p' -> fmap ctx <$> xGrpInfo gInfo' m'' p' msg brokerTs
               XGrpPrefs ps' -> fmap ctx <$> xGrpPrefs gInfo' m'' ps' msg
-              XGrpRoster gr -> fmap ctx <$> xGrpRoster gInfo' m'' gr verifiedMsg sharedMsgId_ brokerTs
+              XGrpRoster gr -> fmap ctx <$> xGrpRoster gInfo' m'' m'' gr verifiedMsg sharedMsgId_ brokerTs
               XGrpRosterAck ackVer ackErr -> Nothing <$ xGrpRosterAck gInfo' m'' ackVer ackErr
               -- TODO [knocking] why don't we forward these messages?
               XGrpDirectInv connReq mContent_ msgScope -> memberCanSend (Just m'') msgScope $ Nothing <$ xGrpDirectInv gInfo' m'' conn' connReq mContent_ msg brokerTs
@@ -1092,7 +1092,7 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage =
               XInfoProbe probe -> Nothing <$ xInfoProbe (COMGroupMember m'') probe
               XInfoProbeCheck probeHash -> Nothing <$ xInfoProbeCheck (COMGroupMember m'') probeHash
               XInfoProbeOk probe -> Nothing <$ xInfoProbeOk (COMGroupMember m'') probe
-              BFileChunk sharedMsgId chunk -> Nothing <$ bFileChunkGroup gInfo' sharedMsgId chunk msgMeta
+              BFileChunk sharedMsgId chunk -> Nothing <$ bFileChunkGroup gInfo' m'' sharedMsgId chunk msgMeta
               _ -> Nothing <$ messageError ("unsupported message: " <> tshow event)
             forM deliveryTaskContext_ $ \taskContext ->
               pure $ NewMessageDeliveryTask {messageId = msgId, taskContext}
@@ -1318,7 +1318,10 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage =
     receiveFileChunk :: Maybe GroupInfo -> RcvFileTransfer -> Maybe Connection -> MsgMeta -> FileChunk -> CM ()
     receiveFileChunk gInfo_ ft@RcvFileTransfer {fileId, fileType, chunkSize} conn_ MsgMeta {recipient = (msgId, _), integrity} = \case
       FileChunkCancel -> case fileType of
-        FTRoster -> forM_ gInfo_ $ cleanupGroupRosterFile user
+        -- cancel only this source's transfer; other relays' in-flight transfers are independent
+        FTRoster -> do
+          t_ <- withStore' $ \db -> getRosterTransfer db fileId
+          forM_ t_ $ \RcvRosterTransfer {rosterTransferId} -> cleanupRosterTransferById rosterTransferId
         FTNormal ->
           unless (rcvFileCompleteOrCancelled ft) $ do
             cancelRcvFileTransfer user ft
@@ -2478,9 +2481,9 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage =
     -- A group BFileChunk is a normal inline file chunk or a roster blob chunk, both located by
     -- (group_id, shared_msg_id). A chunk matching no in-flight transfer (an orphaned re-served roster
     -- chunk, or a missing normal file) is ignored; the outer withAckMessage acks it.
-    bFileChunkGroup :: GroupInfo -> SharedMsgId -> FileChunk -> MsgMeta -> CM ()
-    bFileChunkGroup gInfo@GroupInfo {groupId} sharedMsgId chunk meta = do
-      fileId_ <- withStore' $ \db -> getGroupRcvFileIdBySharedMsgId db userId groupId sharedMsgId
+    bFileChunkGroup :: GroupInfo -> GroupMember -> SharedMsgId -> FileChunk -> MsgMeta -> CM ()
+    bFileChunkGroup gInfo@GroupInfo {groupId} fromMember sharedMsgId chunk meta = do
+      fileId_ <- withStore' $ \db -> getGroupRcvFileId db userId groupId (groupMemberId' fromMember) sharedMsgId
       forM_ fileId_ $ \fileId -> do
         ft <- withStore $ \db -> getRcvFileTransfer db user fileId
         case fileType ft of
@@ -3265,8 +3268,10 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage =
 
     -- The header only starts the transfer; the roster is applied and the version bumped only at
     -- blob completion, so a withheld or corrupted blob leaves the last good roster intact.
-    xGrpRoster :: GroupInfo -> GroupMember -> GroupRoster -> VerifiedMsg e -> Maybe SharedMsgId -> UTCTime -> CM (Maybe DeliveryJobScope)
-    xGrpRoster gInfo author GroupRoster {version = newVer, fileInv = InlineFileInvitation {fileSize, fileDigest}} verifiedMsg sharedMsgId_ brokerTs
+    -- fromMember is the relay that delivered THIS roster copy (the owner on a relay receiving directly,
+    -- a relay on a member receiving a forward); author is the owner who signed it.
+    xGrpRoster :: GroupInfo -> GroupMember -> GroupMember -> GroupRoster -> VerifiedMsg e -> Maybe SharedMsgId -> UTCTime -> CM (Maybe DeliveryJobScope)
+    xGrpRoster gInfo fromMember author GroupRoster {version = newVer, fileInv = InlineFileInvitation {fileSize, fileDigest}} verifiedMsg sharedMsgId_ brokerTs
       -- only an owner may sign a roster; otherwise a relay could route it as a member whose key it controls
       | memberRole' author /= GROwner = messageError "x.grp.roster: not signed by an owner" $> Nothing
       | fileSize > maxGroupRosterBytes = messageError "x.grp.roster: roster blob size exceeds limit" $> Nothing
@@ -3276,68 +3281,73 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage =
           VMSigned _ sm _ -> case sharedMsgId_ of
             Nothing -> Nothing <$ messageWarning "x.grp.roster: missing shared message id"
             Just sharedMsgId -> do
-              pendingVer_ <- fmap (\(v, _, _, _, _) -> v) <$> withStore' (\db -> getRosterPending db gInfo)
-              -- accept a version not below BOTH applied and pending (>=, Nothing below 0): a preceding signed
-              -- event may have already advanced rosterVersion to this blob's version; a lower one is a downgrade.
-              if newVer `aboveRoster` rosterVersion gInfo && newVer `aboveRoster` pendingVer_
+              -- per-source pending version (THIS relay's own in-flight transfer), not a single group slot
+              pendingVer_ <- withStore' $ \db -> getRosterTransferVersion db gInfo (groupMemberId' fromMember)
+              -- accept a version not below BOTH applied and this source's pending (>=, Nothing below 0): a preceding
+              -- signed event may have already advanced rosterVersion to this blob's version; a lower one is a downgrade.
+              if newVer `notBelowRoster` rosterVersion gInfo && newVer `notBelowRoster` pendingVer_
                 then startRosterTransfer sm sharedMsgId
                 else pure Nothing
       where
         startRosterTransfer sm sharedMsgId = do
-          -- supersede any in-flight roster file (older version or a restart) before the new transfer
-          cleanupGroupRosterFile user gInfo
+          -- supersede THIS source's own in-flight transfer (older version or a restart); other relays' transfers are independent
+          cleanupRosterTransfer gInfo (groupMemberId' fromMember)
           let relayHdr = if isUserGrpFwdRelay gInfo then Just sm else Nothing
-          withStore' $ \db -> setRosterPending db gInfo newVer fileDigest (groupMemberId' author) brokerTs relayHdr
           chSize <- asks $ fileChunkSize . config
           let rosterFInv = FileInvitation {fileName = "roster", fileSize, fileDigest = Nothing, fileConnReq = Nothing, fileInline = Just IFMSent, fileDescr = Nothing}
-          rft@RcvFileTransfer {fileId} <- withStore $ \db -> createRcvGroupFileTransfer db userId gInfo Nothing FTRoster (Just sharedMsgId) rosterFInv (Just IFMSent) (fromIntegral chSize)
+          -- transfer record + its scratch file in one transaction (file owned by the transfer, keyed per source)
+          rft@RcvFileTransfer {fileId} <- withStore $ \db -> do
+            transferId <- liftIO $ createRosterTransfer db gInfo (groupMemberId' fromMember) newVer fileDigest (groupMemberId' author) brokerTs relayHdr
+            createRosterRcvFile db userId gInfo fromMember transferId sharedMsgId rosterFInv (Just IFMSent) (fromIntegral chSize)
           -- accept the chat-item-free file before chunk 1 (FIFO before it) so chunk 1 isn't rejected on RFSNew
           -- transient scratch file (consumed into roster_blob, then deleted): temp folder, not the user's files folder / Downloads
           tmpDir <- lift getChatTempDirectory
           rosterTs <- liftIO getCurrentTime
           let GroupInfo {groupId = gId} = gInfo
-              rosterFile = "roster_" <> show gId <> "_" <> formatTime defaultTimeLocale "%Y%m%d_%H%M%S" rosterTs
+              rosterFile = "roster_" <> show gId <> "_" <> show (groupMemberId' fromMember) <> "_" <> formatTime defaultTimeLocale "%Y%m%d_%H%M%S" rosterTs
           filePath <- getRcvFilePath fileId (Just tmpDir) rosterFile False
           withStore' $ \db -> startRcvInlineFT db user rft filePath (Just IFMSent)
           pure Nothing
 
     -- Roster version comparison treating Nothing (un-materialized) as below 0. Non-strict (>=) so a relay
     -- accepts the owner's blob at the version a preceding signed event already advanced rosterVersion to.
-    aboveRoster :: VersionRoster -> Maybe VersionRoster -> Bool
-    aboveRoster v = maybe True (v >=)
+    notBelowRoster :: VersionRoster -> Maybe VersionRoster -> Bool
+    notBelowRoster v = maybe True (v >=)
 
     -- Blob arrived: verify the owner-attested digest over the plaintext and guard against
     -- downgrade before applying; on a relay, ack the owner and re-serve to members.
     rosterCompletion :: GroupInfo -> RcvFileTransfer -> CM ()
-    rosterCompletion gInfo RcvFileTransfer {fileStatus} =
-      withStore' (\db -> getRosterPending db gInfo) >>= \case
-        Nothing -> cleanupGroupRosterFile user gInfo
-        Just (pendingVer, pendingDigest, ownerGMId, rosterBrokerTs, _) -> do
+    rosterCompletion gInfo RcvFileTransfer {fileId, fileStatus} =
+      withStore' (\db -> getRosterTransfer db fileId) >>= \case
+        -- defensive: the file always has its transfer (created together, deleted together)
+        Nothing -> lift (closeFileHandle fileId rcvFiles) >> forM_ (rosterFilePath fileStatus) removeFsFile
+        Just RcvRosterTransfer {rosterTransferId = transferId, rosterTransferVersion = pendingVer, rosterTransferDigest = pendingDigest, rosterTransferOwnerGMId = ownerGMId, rosterTransferBrokerTs = rosterBrokerTs, rosterTransferHeader = header_} -> do
           owner_ <- withStore' $ \db -> eitherToMaybe <$> runExceptT (getGroupMemberById db cxt user ownerGMId)
           blob <- readAssembledRoster
           let isRelay = isUserGrpFwdRelay gInfo
               ackErr err = do
-                cleanupGroupRosterFile user gInfo
+                cleanupRosterTransferById transferId
                 when isRelay $ forM_ owner_ $ \owner -> sendRosterAck gInfo owner pendingVer (Just err)
           if FD.FileDigest (LC.sha512Hash (LB.fromStrict blob)) /= pendingDigest
             then ackErr "relay could not verify the roster blob"
             else case parseAll rosterBlobP blob of
               Left _ -> ackErr "relay could not parse the roster blob"
               Right entries -> case owner_ of
-                Nothing -> cleanupGroupRosterFile user gInfo
+                Nothing -> cleanupRosterTransferById transferId
                 Just author -> do
                   defaultRole <- unknownMemberRole gInfo
-                  -- gate against the persisted roster_version inside the apply transaction, so a reordered
-                  -- same-batch event cannot make this completion downgrade it; stale completion is rejected.
+                  -- gate against the persisted roster_version inside the apply transaction: a roster from another
+                  -- relay (or a preceding signed event) may already have advanced it past this one; a stale
+                  -- completion (e.g. relay1 sent v5 then v6, relay2's v5 completes after v6) is rejected.
                   results_ <- withStore $ \db -> do
                     cur <- liftIO $ getGroupRosterVersion db gInfo
                     if maybe False (pendingVer <) cur
                       then pure Nothing
                       else do
                         res <- processRosterEntries db gInfo defaultRole (validateGroupRoster entries)
-                        liftIO $ promoteRosterPending db gInfo blob
+                        liftIO $ setGroupLiveRoster db gInfo pendingVer ownerGMId rosterBrokerTs header_ blob
                         pure (Just res)
-                  cleanupGroupRosterFile user gInfo
+                  cleanupRosterTransferById transferId
                   forM_ results_ $ \results -> do
                     emitRosterResults gInfo author rosterBrokerTs results
                     -- ack while setting up (own status accepted/acknowledged); a serving (active) relay must not ack broadcasts.
@@ -3345,11 +3355,14 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage =
                       sendRosterAck gInfo author pendingVer Nothing
                       withStore' $ \db -> void $ updateRelayOwnStatusFromTo db gInfo RSAccepted RSAcknowledgedRoster
       where
-        readAssembledRoster = case fileStatus of
-          RFSAccepted fp -> readAt fp
-          RFSConnected fp -> readAt fp
-          RFSComplete fp -> readAt fp
-          _ -> throwChatError $ CEInternalError "roster file not in progress"
+        rosterFilePath = \case
+          RFSAccepted p -> Just p
+          RFSConnected p -> Just p
+          RFSComplete p -> Just p
+          _ -> Nothing
+        readAssembledRoster = case rosterFilePath fileStatus of
+          Just fp -> readAt fp
+          Nothing -> throwChatError $ CEInternalError "roster file not in progress"
         readAt fp = lift (toFSFilePath fp) >>= liftIO . B.readFile
 
     -- TOFU-apply an owner-signed (key, role) to a resolved member: pin the key if absent; for a keyed
@@ -3755,7 +3768,7 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage =
             XGrpDel -> withAuthor XGrpDel_ $ \author -> void $ xGrpDel gInfo author rcvMsg msgTs
             XGrpInfo p' -> withAuthor XGrpInfo_ $ \author -> void $ xGrpInfo gInfo author p' rcvMsg msgTs
             XGrpPrefs ps' -> withAuthor XGrpPrefs_ $ \author -> void $ xGrpPrefs gInfo author ps' rcvMsg
-            XGrpRoster gr -> withAuthor XGrpRoster_ $ \author -> void $ xGrpRoster gInfo author gr verifiedMsg sharedMsgId_ msgTs
+            XGrpRoster gr -> withAuthor XGrpRoster_ $ \author -> void $ xGrpRoster gInfo m author gr verifiedMsg sharedMsgId_ msgTs
             _ -> messageError $ "x.grp.msg.forward: unsupported forwarded event " <> T.pack (show $ toCMEventTag event)
           where
             withAuthor :: CMEventTag e -> (GroupMember -> CM ()) -> CM ()

src/Simplex/Chat/Store/Files.hs

@@ -31,9 +31,11 @@ module Simplex.Chat.Store.Files
     getSharedMsgIdByFileId,
     getFileIdBySharedMsgId,
     getGroupFileIdBySharedMsgId,
-    getGroupRcvFileIdBySharedMsgId,
+    getGroupRcvFileId,
     getGroupRosterFileInfo,
     deleteGroupRosterFile,
+    getRosterTransferFile,
+    deleteRosterTransferFile,
     getRcvFileLastChunkNo,
     getDirectFileIdBySharedMsgId,
     getChatRefByFileId,
@@ -41,6 +43,7 @@ module Simplex.Chat.Store.Files
     updateSndFileStatus,
     createRcvFileTransfer,
     createRcvGroupFileTransfer,
+    createRosterRcvFile,
     createRcvStandaloneFileTransfer,
     appendRcvFD,
     getRcvFileDescrByRcvFileId,
@@ -325,20 +328,42 @@ getGroupFileIdBySharedMsgId db userId groupId sharedMsgId =
       |]
       (userId, groupId, sharedMsgId)
 
--- Locates a received group file (normal or roster) by the header's shared_msg_id without the
--- chat_items join, so it finds the chat-item-free roster blob and normal inline files alike.
--- Nothing => no in-flight transfer (an orphaned re-served chunk), which the caller ACKs and ignores.
-getGroupRcvFileIdBySharedMsgId :: DB.Connection -> UserId -> Int64 -> SharedMsgId -> IO (Maybe Int64)
-getGroupRcvFileIdBySharedMsgId db userId groupId sharedMsgId =
-  maybeFirstRow fromOnly $
-    DB.query
-      db
+-- Resolve the in-flight received group inline file for a chunk: read its file_type by shared_msg_id
+-- (LIMIT 1 is safe -- all files sharing a shared_msg_id share a type), then look up by type: a roster
+-- file is scoped to its source relay (every relay re-serves the owner's same shared_msg_id, so the source
+-- disambiguates), a normal file is by shared_msg_id. Nothing => no in-flight transfer (orphaned chunk).
+getGroupRcvFileId :: DB.Connection -> UserId -> Int64 -> GroupMemberId -> SharedMsgId -> IO (Maybe Int64)
+getGroupRcvFileId db userId groupId fromMemberId sharedMsgId = do
+  fileType_ <- getFileType
+  case fileType_ of
+    Just FTRoster ->
+      maybeFirstRow fromOnly $
+        DB.query db (rcvFileIdQ <> " AND r.group_member_id = ?") (userId, groupId, sharedMsgId, FTRoster, fromMemberId)
+    Just FTNormal ->
+      maybeFirstRow fromOnly $
+        DB.query db rcvFileIdQ (userId, groupId, sharedMsgId, FTNormal)
+    Nothing -> pure Nothing
+  where
+    getFileType =
+      maybeFirstRow fromOnly $
+        DB.query db "SELECT file_type FROM files WHERE user_id = ? AND group_id = ? AND shared_msg_id = ? LIMIT 1" (userId, groupId, sharedMsgId)
+    rcvFileIdQ =
       [sql|
         SELECT f.file_id FROM files f
         JOIN rcv_files r ON r.file_id = f.file_id
-        WHERE f.user_id = ? AND f.group_id = ? AND f.shared_msg_id = ?
+        WHERE f.user_id = ? AND f.group_id = ? AND f.shared_msg_id = ? AND f.file_type = ?
       |]
-      (userId, groupId, sharedMsgId)
+
+-- The roster scratch file for a transfer (for fs/handle cleanup before deleting the transfer).
+-- A transfer owns exactly one file (created together in one transaction), so this is single-valued.
+getRosterTransferFile :: DB.Connection -> Int64 -> IO (Maybe (Int64, Maybe FilePath))
+getRosterTransferFile db transferId =
+  maybeFirstRow id $ DB.query db "SELECT file_id, file_path FROM files WHERE roster_transfer_id = ?" (Only transferId)
+
+-- Deletes a transfer's file row; rcv_files and rcv_file_chunks cascade on the FK.
+deleteRosterTransferFile :: DB.Connection -> Int64 -> IO ()
+deleteRosterTransferFile db transferId =
+  DB.execute db "DELETE FROM files WHERE roster_transfer_id = ?" (Only transferId)
 
 -- For roster-file cleanup keyed on the group (not a chat item): every matching file_id and its on-disk
 -- path, so the caller evicts the handle and removes the file for each — delete-all like deleteGroupRosterFile.
@@ -444,6 +469,25 @@ createRcvGroupFileTransfer db userId GroupInfo {groupId, localDisplayName = gNam
       (fileId, FSNew, fileConnReq, fileInline, rcvFileInline, grpMemberId_, rfdId, currentTs, currentTs)
   pure RcvFileTransfer {fileId, xftpRcvFile, fileInvitation = f, fileStatus = RFSNew, fileType, rcvFileInline, senderDisplayName = senderName, chunkSize, cancelled = False, grpMemberId = grpMemberId_, cryptoArgs = Nothing}
 
+-- Roster scratch file owned by a per-source transfer: group_member_id is the delivering relay (so chunk
+-- streams from different relays are distinct files), roster_transfer_id links to the metadata record.
+createRosterRcvFile :: DB.Connection -> UserId -> GroupInfo -> GroupMember -> Int64 -> SharedMsgId -> FileInvitation -> Maybe InlineFileMode -> Integer -> ExceptT StoreError IO RcvFileTransfer
+createRosterRcvFile db userId GroupInfo {groupId} src@GroupMember {localDisplayName = senderName} transferId sharedMsgId f@FileInvitation {fileName, fileSize, fileConnReq, fileInline} rcvFileInline chunkSize = do
+  currentTs <- liftIO getCurrentTime
+  let grpMemberId_ = groupMemberId' src
+  fileId <- liftIO $ do
+    DB.execute
+      db
+      "INSERT INTO files (user_id, group_id, file_name, file_size, chunk_size, file_inline, ci_file_status, protocol, file_type, shared_msg_id, roster_transfer_id, created_at, updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)"
+      ((userId, groupId, fileName, fileSize, chunkSize, fileInline, CIFSRcvInvitation, FPSMP, FTRoster) :. (sharedMsgId, transferId, currentTs, currentTs))
+    insertedRowId db
+  liftIO $
+    DB.execute
+      db
+      "INSERT INTO rcv_files (file_id, file_status, file_queue_info, file_inline, rcv_file_inline, group_member_id, created_at, updated_at) VALUES (?,?,?,?,?,?,?,?)"
+      (fileId, FSNew, fileConnReq, fileInline, rcvFileInline, grpMemberId_, currentTs, currentTs)
+  pure RcvFileTransfer {fileId, xftpRcvFile = Nothing, fileInvitation = f, fileStatus = RFSNew, fileType = FTRoster, rcvFileInline, senderDisplayName = senderName, chunkSize, cancelled = False, grpMemberId = Just grpMemberId_, cryptoArgs = Nothing}
+
 createRcvStandaloneFileTransfer :: DB.Connection -> UserId -> CryptoFile -> Int64 -> Word32 -> ExceptT StoreError IO Int64
 createRcvStandaloneFileTransfer db userId (CryptoFile filePath cfArgs_) fileSize chunkSize = do
   currentTs <- liftIO getCurrentTime

src/Simplex/Chat/Store/Groups.hs

@@ -91,10 +91,14 @@ module Simplex.Chat.Store.Groups
     setGroupRosterVersion,
     getGroupRosterVersion,
     getGroupRoster,
-    setRosterPending,
-    getRosterPending,
-    promoteRosterPending,
-    clearRosterPending,
+    RcvRosterTransfer (..),
+    createRosterTransfer,
+    getRosterTransferVersion,
+    getRosterTransferId,
+    getRosterTransfer,
+    setGroupLiveRoster,
+    deleteRosterTransfer,
+    deleteGroupRosterTransfers,
     setGroupMemberKeyRole,
     createRelayForOwner,
     getCreateRelayForMember,
@@ -1479,90 +1483,115 @@ getGroupRoster db GroupInfo {groupId} =
       (\sigs -> (ownerGMId, brokerTs, SignedMsg cb sigs body, (\(Binary b) -> b) <$> blob_)) <$> eitherToMaybe (smpDecode sigsBs)
     toRoster _ = Nothing
 
--- The signed-header columns are relay-only (Nothing on a member); promoted to the live header at
--- completion so the relay can re-forward the roster.
-setRosterPending :: DB.Connection -> GroupInfo -> VersionRoster -> FD.FileDigest -> GroupMemberId -> UTCTime -> Maybe SignedMsg -> IO ()
-setRosterPending db GroupInfo {groupId} v digest ownerGMId brokerTs sm_ = do
-  currentTs <- getCurrentTime
+-- A per-source in-flight roster transfer, keyed (group_id, from_member_id): replaces the single
+-- roster_pending_* slot, so two relays serving one member can't share a chunk stream. The signed-header
+-- columns are relay-only (NULL on members), promoted to the live roster_msg_* on groups at completion.
+createRosterTransfer :: DB.Connection -> GroupInfo -> GroupMemberId -> VersionRoster -> FD.FileDigest -> GroupMemberId -> UTCTime -> Maybe SignedMsg -> IO Int64
+createRosterTransfer db GroupInfo {groupId} fromMemberId v digest ownerGMId brokerTs sm_ = do
+  -- one in-flight transfer per (group, source): drop any prior row from this source so the INSERT can't hit
+  -- the UNIQUE constraint even if the caller's fs/handle cleanup was skipped (the scratch file would then leak
+  -- until group delete, but the transfer never gets stuck). Normally cleanupRosterTransfer ran first.
+  DB.execute db "DELETE FROM rcv_roster_transfers WHERE group_id = ? AND from_member_id = ?" (groupId, fromMemberId)
   DB.execute
     db
     [sql|
-      UPDATE groups
-      SET roster_pending_version = ?, roster_pending_digest = ?,
-          roster_pending_sending_owner_gm_id = ?, roster_pending_broker_ts = ?,
-          roster_pending_msg_chat_binding = ?, roster_pending_msg_signatures = ?, roster_pending_msg_body = ?,
-          updated_at = ?
-      WHERE group_id = ?
+      INSERT INTO rcv_roster_transfers
+        (group_id, from_member_id, roster_version, roster_digest, sending_owner_gm_id, broker_ts,
+         roster_msg_chat_binding, roster_msg_signatures, roster_msg_body)
+      VALUES (?,?,?,?,?,?,?,?,?)
     |]
-    ((v, Binary (FD.unFileDigest digest), ownerGMId, brokerTs)
-       :. ((\SignedMsg {chatBinding} -> chatBinding) <$> sm_, (\SignedMsg {signatures} -> Binary (smpEncode signatures)) <$> sm_, (\SignedMsg {signedBody} -> Binary signedBody) <$> sm_, currentTs, groupId))
+    ( (groupId, fromMemberId, v, Binary (FD.unFileDigest digest), ownerGMId, brokerTs)
+        :. ((\SignedMsg {chatBinding} -> chatBinding) <$> sm_, (\SignedMsg {signatures} -> Binary (smpEncode signatures)) <$> sm_, (\SignedMsg {signedBody} -> Binary signedBody) <$> sm_)
+    )
+  insertedRowId db
 
-getRosterPending :: DB.Connection -> GroupInfo -> IO (Maybe (VersionRoster, FD.FileDigest, GroupMemberId, UTCTime, Maybe SignedMsg))
-getRosterPending db GroupInfo {groupId} =
-  (>>= toPending)
+getRosterTransferVersion :: DB.Connection -> GroupInfo -> GroupMemberId -> IO (Maybe VersionRoster)
+getRosterTransferVersion db GroupInfo {groupId} fromMemberId =
+  maybeFirstRow fromOnly $
+    DB.query db "SELECT roster_version FROM rcv_roster_transfers WHERE group_id = ? AND from_member_id = ?" (groupId, fromMemberId)
+
+getRosterTransferId :: DB.Connection -> GroupInfo -> GroupMemberId -> IO (Maybe Int64)
+getRosterTransferId db GroupInfo {groupId} fromMemberId =
+  maybeFirstRow fromOnly $
+    DB.query db "SELECT roster_transfer_id FROM rcv_roster_transfers WHERE group_id = ? AND from_member_id = ?" (groupId, fromMemberId)
+
+-- An in-flight received roster transfer (a rcv_roster_transfers row joined to its scratch file), read at
+-- completion. The header is the relay's re-serve SignedMsg -- present only on a serving relay (NULL on a
+-- member, whose live roster_msg_* stay NULL so it never re-serves).
+data RcvRosterTransfer = RcvRosterTransfer
+  { rosterTransferId :: Int64,
+    rosterTransferVersion :: VersionRoster,
+    rosterTransferDigest :: FD.FileDigest,
+    rosterTransferOwnerGMId :: GroupMemberId,
+    rosterTransferBrokerTs :: UTCTime,
+    rosterTransferHeader :: Maybe SignedMsg
+  }
+  deriving (Show)
+
+-- The in-flight transfer for a received roster file (joined via files.roster_transfer_id), with its
+-- relay-only signed header. Read at completion to apply, promote into the live roster, and ack.
+getRosterTransfer :: DB.Connection -> Int64 -> IO (Maybe RcvRosterTransfer)
+getRosterTransfer db fileId =
+  (>>= toTransfer)
     <$> maybeFirstRow
       id
       ( DB.query
           db
           [sql|
-            SELECT roster_pending_version, roster_pending_digest,
-                   roster_pending_sending_owner_gm_id, roster_pending_broker_ts,
-                   roster_pending_msg_chat_binding, roster_pending_msg_signatures, roster_pending_msg_body
-            FROM groups WHERE group_id = ?
+            SELECT t.roster_transfer_id, t.roster_version, t.roster_digest, t.sending_owner_gm_id, t.broker_ts,
+                   t.roster_msg_chat_binding, t.roster_msg_signatures, t.roster_msg_body
+            FROM rcv_roster_transfers t
+            JOIN files f ON f.roster_transfer_id = t.roster_transfer_id
+            WHERE f.file_id = ?
           |]
-          (Only groupId)
+          (Only fileId)
       )
   where
-    toPending (Just v, Just (Binary d), Just ownerGMId, Just brokerTs, cb_, sigs_, body_) =
-      Just (v, FD.FileDigest d, ownerGMId, brokerTs, sm_)
+    toTransfer (tId, v, Binary d, ownerGMId, brokerTs, cb_, sigs_, body_) =
+      Just
+        RcvRosterTransfer
+          { rosterTransferId = tId,
+            rosterTransferVersion = v,
+            rosterTransferDigest = FD.FileDigest d,
+            rosterTransferOwnerGMId = ownerGMId,
+            rosterTransferBrokerTs = brokerTs,
+            rosterTransferHeader = sm_
+          }
       where
         sm_ = case (cb_, sigs_, body_) of
           (Just cb, Just (Binary sigsBs), Just (Binary body)) ->
             (\sigs -> SignedMsg cb sigs body) <$> eitherToMaybe (smpDecode sigsBs)
           _ -> Nothing
-    toPending _ = Nothing
 
--- Promote pending -> live (header + blob + version) and clear pending, in one statement.
-promoteRosterPending :: DB.Connection -> GroupInfo -> ByteString -> IO ()
-promoteRosterPending db GroupInfo {groupId} blob = do
+-- Write the single live roster on groups from a completed transfer's values (header NULL on a member,
+-- so its live roster_msg_* stay NULL and it never re-serves; only relays re-serve).
+setGroupLiveRoster :: DB.Connection -> GroupInfo -> VersionRoster -> GroupMemberId -> UTCTime -> Maybe SignedMsg -> ByteString -> IO ()
+setGroupLiveRoster db GroupInfo {groupId} v ownerGMId brokerTs sm_ blob = do
   currentTs <- getCurrentTime
   DB.execute
     db
     [sql|
       UPDATE groups SET
-        roster_version = roster_pending_version,
-        roster_blob = ?,
-        roster_sending_owner_gm_id = roster_pending_sending_owner_gm_id,
-        roster_broker_ts = roster_pending_broker_ts,
-        roster_msg_chat_binding = roster_pending_msg_chat_binding,
-        roster_msg_signatures = roster_pending_msg_signatures,
-        roster_msg_body = roster_pending_msg_body,
-        roster_pending_version = NULL,
-        roster_pending_digest = NULL,
-        roster_pending_sending_owner_gm_id = NULL,
-        roster_pending_broker_ts = NULL,
-        roster_pending_msg_chat_binding = NULL,
-        roster_pending_msg_signatures = NULL,
-        roster_pending_msg_body = NULL,
+        roster_version = ?, roster_blob = ?,
+        roster_sending_owner_gm_id = ?, roster_broker_ts = ?,
+        roster_msg_chat_binding = ?, roster_msg_signatures = ?, roster_msg_body = ?,
         updated_at = ?
       WHERE group_id = ?
     |]
-    (Binary blob, currentTs, groupId)
+    ( (v, Binary blob, ownerGMId, brokerTs)
+        :. ((\SignedMsg {chatBinding} -> chatBinding) <$> sm_, (\SignedMsg {signatures} -> Binary (smpEncode signatures)) <$> sm_, (\SignedMsg {signedBody} -> Binary signedBody) <$> sm_, currentTs, groupId)
+    )
 
-clearRosterPending :: DB.Connection -> GroupInfo -> IO ()
-clearRosterPending db GroupInfo {groupId} = do
-  currentTs <- getCurrentTime
-  DB.execute
-    db
-    [sql|
-      UPDATE groups SET
-        roster_pending_version = NULL, roster_pending_digest = NULL,
-        roster_pending_sending_owner_gm_id = NULL, roster_pending_broker_ts = NULL,
-        roster_pending_msg_chat_binding = NULL, roster_pending_msg_signatures = NULL, roster_pending_msg_body = NULL,
-        updated_at = ?
-      WHERE group_id = ?
-    |]
-    (currentTs, groupId)
+-- Delete one in-flight transfer row (its files/rcv_files/rcv_file_chunks are removed separately, with
+-- the on-disk file). Caller removes the fs file + cached handle first.
+deleteRosterTransfer :: DB.Connection -> Int64 -> IO ()
+deleteRosterTransfer db transferId =
+  DB.execute db "DELETE FROM rcv_roster_transfers WHERE roster_transfer_id = ?" (Only transferId)
+
+-- All in-flight transfers for a group (group delete).
+deleteGroupRosterTransfers :: DB.Connection -> Int64 -> IO ()
+deleteGroupRosterTransfers db groupId =
+  DB.execute db "DELETE FROM rcv_roster_transfers WHERE group_id = ?" (Only groupId)
 
 setGroupMemberKeyRole :: DB.Connection -> GroupMember -> C.PublicKeyEd25519 -> GroupMemberRole -> IO ()
 setGroupMemberKeyRole db GroupMember {groupMemberId} pubKey role = do

src/Simplex/Chat/Store/Postgres/Migrations/M20260601_group_roster.hs

@@ -16,31 +16,44 @@ ALTER TABLE groups ADD COLUMN roster_msg_signatures BYTEA;
 ALTER TABLE groups ADD COLUMN roster_sending_owner_gm_id BIGINT;
 ALTER TABLE groups ADD COLUMN roster_broker_ts TIMESTAMPTZ;
 ALTER TABLE groups ADD COLUMN roster_blob BYTEA;
-ALTER TABLE groups ADD COLUMN roster_pending_version BIGINT;
-ALTER TABLE groups ADD COLUMN roster_pending_digest BYTEA;
-ALTER TABLE groups ADD COLUMN roster_pending_msg_body BYTEA;
-ALTER TABLE groups ADD COLUMN roster_pending_msg_chat_binding TEXT;
-ALTER TABLE groups ADD COLUMN roster_pending_msg_signatures BYTEA;
-ALTER TABLE groups ADD COLUMN roster_pending_sending_owner_gm_id BIGINT;
-ALTER TABLE groups ADD COLUMN roster_pending_broker_ts TIMESTAMPTZ;
+
+CREATE TABLE rcv_roster_transfers(
+  roster_transfer_id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
+  group_id BIGINT NOT NULL REFERENCES groups ON DELETE CASCADE,
+  from_member_id BIGINT NOT NULL REFERENCES group_members ON DELETE CASCADE,
+  roster_version BIGINT NOT NULL,
+  roster_digest BYTEA NOT NULL,
+  sending_owner_gm_id BIGINT NOT NULL,
+  broker_ts TIMESTAMPTZ NOT NULL,
+  roster_msg_body BYTEA,
+  roster_msg_chat_binding TEXT,
+  roster_msg_signatures BYTEA,
+  created_at TEXT NOT NULL DEFAULT (now()),
+  updated_at TEXT NOT NULL DEFAULT (now())
+);
+CREATE UNIQUE INDEX idx_rcv_roster_transfers_group_id_from_member_id ON rcv_roster_transfers(group_id, from_member_id);
+CREATE INDEX idx_rcv_roster_transfers_from_member_id ON rcv_roster_transfers(from_member_id);
+
 ALTER TABLE files ADD COLUMN shared_msg_id BYTEA;
 ALTER TABLE files ADD COLUMN file_type TEXT NOT NULL DEFAULT 'normal';
+ALTER TABLE files ADD COLUMN roster_transfer_id BIGINT;
 CREATE INDEX idx_files_group_id_shared_msg_id ON files(group_id, shared_msg_id);
+CREATE INDEX idx_files_roster_transfer_id ON files(roster_transfer_id);
 |]
 
 down_m20260601_group_roster :: Text
 down_m20260601_group_roster =
   [r|
-DROP INDEX IF EXISTS idx_files_group_id_shared_msg_id;
+DROP INDEX idx_files_roster_transfer_id;
+DROP INDEX idx_files_group_id_shared_msg_id;
+ALTER TABLE files DROP COLUMN roster_transfer_id;
 ALTER TABLE files DROP COLUMN file_type;
 ALTER TABLE files DROP COLUMN shared_msg_id;
-ALTER TABLE groups DROP COLUMN roster_pending_broker_ts;
-ALTER TABLE groups DROP COLUMN roster_pending_sending_owner_gm_id;
-ALTER TABLE groups DROP COLUMN roster_pending_msg_signatures;
-ALTER TABLE groups DROP COLUMN roster_pending_msg_chat_binding;
-ALTER TABLE groups DROP COLUMN roster_pending_msg_body;
-ALTER TABLE groups DROP COLUMN roster_pending_digest;
-ALTER TABLE groups DROP COLUMN roster_pending_version;
+
+DROP INDEX idx_rcv_roster_transfers_from_member_id;
+DROP INDEX idx_rcv_roster_transfers_group_id_from_member_id;
+DROP TABLE rcv_roster_transfers;
+
 ALTER TABLE groups DROP COLUMN roster_blob;
 ALTER TABLE groups DROP COLUMN roster_broker_ts;
 ALTER TABLE groups DROP COLUMN roster_sending_owner_gm_id;

src/Simplex/Chat/Store/Postgres/Migrations/chat_schema.sql

@@ -754,7 +754,8 @@ CREATE TABLE test_chat_schema.files (
     note_folder_id bigint,
     redirect_file_id bigint,
     shared_msg_id bytea,
-    file_type text DEFAULT 'normal'::text NOT NULL
+    file_type text DEFAULT 'normal'::text NOT NULL,
+    roster_transfer_id bigint
 );
 
 
@@ -987,14 +988,7 @@ CREATE TABLE test_chat_schema.groups (
     roster_msg_signatures bytea,
     roster_sending_owner_gm_id bigint,
     roster_broker_ts timestamp with time zone,
-    roster_blob bytea,
-    roster_pending_version bigint,
-    roster_pending_digest bytea,
-    roster_pending_msg_body bytea,
-    roster_pending_msg_chat_binding text,
-    roster_pending_msg_signatures bytea,
-    roster_pending_sending_owner_gm_id bigint,
-    roster_pending_broker_ts timestamp with time zone
+    roster_blob bytea
 );
 
 
@@ -1221,6 +1215,34 @@ CREATE TABLE test_chat_schema.rcv_files (
 
 
 
+CREATE TABLE test_chat_schema.rcv_roster_transfers (
+    roster_transfer_id bigint NOT NULL,
+    group_id bigint NOT NULL,
+    from_member_id bigint NOT NULL,
+    roster_version bigint NOT NULL,
+    roster_digest bytea NOT NULL,
+    sending_owner_gm_id bigint NOT NULL,
+    broker_ts timestamp with time zone NOT NULL,
+    roster_msg_body bytea,
+    roster_msg_chat_binding text,
+    roster_msg_signatures bytea,
+    created_at text DEFAULT now() NOT NULL,
+    updated_at text DEFAULT now() NOT NULL
+);
+
+
+
+ALTER TABLE test_chat_schema.rcv_roster_transfers ALTER COLUMN roster_transfer_id ADD GENERATED ALWAYS AS IDENTITY (
+    SEQUENCE NAME test_chat_schema.rcv_roster_transfers_roster_transfer_id_seq
+    START WITH 1
+    INCREMENT BY 1
+    NO MINVALUE
+    NO MAXVALUE
+    CACHE 1
+);
+
+
+
 CREATE TABLE test_chat_schema.received_probes (
     received_probe_id bigint NOT NULL,
     contact_id bigint,
@@ -1754,6 +1776,11 @@ ALTER TABLE ONLY test_chat_schema.rcv_files
 
 
 
+ALTER TABLE ONLY test_chat_schema.rcv_roster_transfers
+    ADD CONSTRAINT rcv_roster_transfers_pkey PRIMARY KEY (roster_transfer_id);
+
+
+
 ALTER TABLE ONLY test_chat_schema.received_probes
     ADD CONSTRAINT received_probes_pkey PRIMARY KEY (received_probe_id);
 
@@ -2295,6 +2322,10 @@ CREATE INDEX idx_files_redirect_file_id ON test_chat_schema.files USING btree (r
 
 
 
+CREATE INDEX idx_files_roster_transfer_id ON test_chat_schema.files USING btree (roster_transfer_id);
+
+
+
 CREATE INDEX idx_files_user_id ON test_chat_schema.files USING btree (user_id);
 
 
@@ -2467,6 +2498,14 @@ CREATE INDEX idx_rcv_files_group_member_id ON test_chat_schema.rcv_files USING b
 
 
 
+CREATE INDEX idx_rcv_roster_transfers_from_member_id ON test_chat_schema.rcv_roster_transfers USING btree (from_member_id);
+
+
+
+CREATE UNIQUE INDEX idx_rcv_roster_transfers_group_id_from_member_id ON test_chat_schema.rcv_roster_transfers USING btree (group_id, from_member_id);
+
+
+
 CREATE INDEX idx_received_probes_contact_id ON test_chat_schema.received_probes USING btree (contact_id);
 
 
@@ -3152,6 +3191,16 @@ ALTER TABLE ONLY test_chat_schema.rcv_files
 
 
 
+ALTER TABLE ONLY test_chat_schema.rcv_roster_transfers
+    ADD CONSTRAINT rcv_roster_transfers_from_member_id_fkey FOREIGN KEY (from_member_id) REFERENCES test_chat_schema.group_members(group_member_id) ON DELETE CASCADE;
+
+
+
+ALTER TABLE ONLY test_chat_schema.rcv_roster_transfers
+    ADD CONSTRAINT rcv_roster_transfers_group_id_fkey FOREIGN KEY (group_id) REFERENCES test_chat_schema.groups(group_id) ON DELETE CASCADE;
+
+
+
 ALTER TABLE ONLY test_chat_schema.received_probes
     ADD CONSTRAINT received_probes_contact_id_fkey FOREIGN KEY (contact_id) REFERENCES test_chat_schema.contacts(contact_id) ON DELETE CASCADE;
 

src/Simplex/Chat/Store/SQLite/Migrations/M20260601_group_roster.hs

@@ -15,31 +15,44 @@ ALTER TABLE groups ADD COLUMN roster_msg_signatures BLOB;
 ALTER TABLE groups ADD COLUMN roster_sending_owner_gm_id INTEGER;
 ALTER TABLE groups ADD COLUMN roster_broker_ts TEXT;
 ALTER TABLE groups ADD COLUMN roster_blob BLOB;
-ALTER TABLE groups ADD COLUMN roster_pending_version INTEGER;
-ALTER TABLE groups ADD COLUMN roster_pending_digest BLOB;
-ALTER TABLE groups ADD COLUMN roster_pending_msg_body BLOB;
-ALTER TABLE groups ADD COLUMN roster_pending_msg_chat_binding TEXT;
-ALTER TABLE groups ADD COLUMN roster_pending_msg_signatures BLOB;
-ALTER TABLE groups ADD COLUMN roster_pending_sending_owner_gm_id INTEGER;
-ALTER TABLE groups ADD COLUMN roster_pending_broker_ts TEXT;
+
+CREATE TABLE rcv_roster_transfers(
+  roster_transfer_id INTEGER PRIMARY KEY,
+  group_id INTEGER NOT NULL REFERENCES groups ON DELETE CASCADE,
+  from_member_id INTEGER NOT NULL REFERENCES group_members ON DELETE CASCADE,
+  roster_version INTEGER NOT NULL,
+  roster_digest BLOB NOT NULL,
+  sending_owner_gm_id INTEGER NOT NULL,
+  broker_ts TEXT NOT NULL,
+  roster_msg_body BLOB,
+  roster_msg_chat_binding TEXT,
+  roster_msg_signatures BLOB,
+  created_at TEXT NOT NULL DEFAULT(datetime('now')),
+  updated_at TEXT NOT NULL DEFAULT(datetime('now'))
+) STRICT;
+CREATE UNIQUE INDEX idx_rcv_roster_transfers_group_id_from_member_id ON rcv_roster_transfers(group_id, from_member_id);
+CREATE INDEX idx_rcv_roster_transfers_from_member_id ON rcv_roster_transfers(from_member_id);
+
 ALTER TABLE files ADD COLUMN shared_msg_id BLOB;
 ALTER TABLE files ADD COLUMN file_type TEXT NOT NULL DEFAULT 'normal';
+ALTER TABLE files ADD COLUMN roster_transfer_id INTEGER;
 CREATE INDEX idx_files_group_id_shared_msg_id ON files(group_id, shared_msg_id);
+CREATE INDEX idx_files_roster_transfer_id ON files(roster_transfer_id);
 |]
 
 down_m20260601_group_roster :: Query
 down_m20260601_group_roster =
   [sql|
-DROP INDEX IF EXISTS idx_files_group_id_shared_msg_id;
+DROP INDEX idx_files_roster_transfer_id;
+DROP INDEX idx_files_group_id_shared_msg_id;
+ALTER TABLE files DROP COLUMN roster_transfer_id;
 ALTER TABLE files DROP COLUMN file_type;
 ALTER TABLE files DROP COLUMN shared_msg_id;
-ALTER TABLE groups DROP COLUMN roster_pending_broker_ts;
-ALTER TABLE groups DROP COLUMN roster_pending_sending_owner_gm_id;
-ALTER TABLE groups DROP COLUMN roster_pending_msg_signatures;
-ALTER TABLE groups DROP COLUMN roster_pending_msg_chat_binding;
-ALTER TABLE groups DROP COLUMN roster_pending_msg_body;
-ALTER TABLE groups DROP COLUMN roster_pending_digest;
-ALTER TABLE groups DROP COLUMN roster_pending_version;
+
+DROP INDEX idx_rcv_roster_transfers_from_member_id;
+DROP INDEX idx_rcv_roster_transfers_group_id_from_member_id;
+DROP TABLE rcv_roster_transfers;
+
 ALTER TABLE groups DROP COLUMN roster_blob;
 ALTER TABLE groups DROP COLUMN roster_broker_ts;
 ALTER TABLE groups DROP COLUMN roster_sending_owner_gm_id;

src/Simplex/Chat/Store/SQLite/Migrations/chat_schema.sql

@@ -198,14 +198,7 @@ CREATE TABLE groups(
   roster_msg_signatures BLOB,
   roster_sending_owner_gm_id INTEGER,
   roster_broker_ts TEXT,
-  roster_blob BLOB,
-  roster_pending_version INTEGER,
-  roster_pending_digest BLOB,
-  roster_pending_msg_body BLOB,
-  roster_pending_msg_chat_binding TEXT,
-  roster_pending_msg_signatures BLOB,
-  roster_pending_sending_owner_gm_id INTEGER,
-  roster_pending_broker_ts TEXT, -- received
+  roster_blob BLOB, -- received
   FOREIGN KEY(user_id, local_display_name)
   REFERENCES display_names(user_id, local_display_name)
   ON DELETE CASCADE
@@ -293,7 +286,8 @@ CREATE TABLE files(
   note_folder_id INTEGER DEFAULT NULL REFERENCES note_folders ON DELETE CASCADE,
   redirect_file_id INTEGER REFERENCES files ON DELETE CASCADE,
   shared_msg_id BLOB,
-  file_type TEXT NOT NULL DEFAULT 'normal'
+  file_type TEXT NOT NULL DEFAULT 'normal',
+  roster_transfer_id INTEGER
 ) STRICT;
 CREATE TABLE snd_files(
   file_id INTEGER NOT NULL REFERENCES files ON DELETE CASCADE,
@@ -813,6 +807,20 @@ CREATE TABLE group_relays(
   ,
   base_web_url TEXT
 ) STRICT;
+CREATE TABLE rcv_roster_transfers(
+  roster_transfer_id INTEGER PRIMARY KEY,
+  group_id INTEGER NOT NULL REFERENCES groups ON DELETE CASCADE,
+  from_member_id INTEGER NOT NULL REFERENCES group_members ON DELETE CASCADE,
+  roster_version INTEGER NOT NULL,
+  roster_digest BLOB NOT NULL,
+  sending_owner_gm_id INTEGER NOT NULL,
+  broker_ts TEXT NOT NULL,
+  roster_msg_body BLOB,
+  roster_msg_chat_binding TEXT,
+  roster_msg_signatures BLOB,
+  created_at TEXT NOT NULL DEFAULT(datetime('now')),
+  updated_at TEXT NOT NULL DEFAULT(datetime('now'))
+) STRICT;
 CREATE INDEX contact_profiles_index ON contact_profiles(
   display_name,
   full_name
@@ -1332,10 +1340,18 @@ ON groups(
   relay_request_group_link
 )
 WHERE relay_request_group_link IS NOT NULL;
+CREATE UNIQUE INDEX idx_rcv_roster_transfers_group_id_from_member_id ON rcv_roster_transfers(
+  group_id,
+  from_member_id
+);
+CREATE INDEX idx_rcv_roster_transfers_from_member_id ON rcv_roster_transfers(
+  from_member_id
+);
 CREATE INDEX idx_files_group_id_shared_msg_id ON files(
   group_id,
   shared_msg_id
 );
+CREATE INDEX idx_files_roster_transfer_id ON files(roster_transfer_id);
 CREATE TRIGGER on_group_members_insert_update_summary
 AFTER INSERT ON group_members
 FOR EACH ROW

tests/ChatTests/Groups.hs

@@ -298,6 +298,7 @@ chatGroupTests = do
         it "promoted member enters the roster and can post" testChannelPromotedMemberCanPost
         it "observer cannot post until promoted" testChannelObserverCannotPost
         it "promoted member re-connecting via a new relay is accepted via the roster-pinned key" testChannelPromotedMemberRejoinViaRelay
+        it "2 relays: multi-chunk roster reassembles per source (no stream interleaving)" testChannelRosterMultiRelayMultipart
     describe "channel message operations" $ do
       it "should update channel message" testChannelMessageUpdate
       it "should delete channel message" testChannelMessageDelete
@@ -8746,7 +8747,7 @@ prepareChannel2Relays gName owner relay1 relay2 = do
         owner <## ("#" <> gName <> ": group link relays updated, current relays:")
         owner
           <### [ EndsWith ": active",
-                 Predicate (\l -> ": invited" `isSuffixOf` l || ": accepted" `isSuffixOf` l)
+                 Predicate (\l -> ": invited" `isSuffixOf` l || ": accepted" `isSuffixOf` l || ": acknowledged_roster" `isSuffixOf` l)
                ]
         owner <## "group link:"
         void $ getTermLine owner -- consume group link line
@@ -10539,7 +10540,7 @@ testChannelRosterDigestMismatchRejected ps =
           -- frank joins; bob re-serves the valid header with the corrupted blob, frank rejects it
           threadDelay 100000
           memberJoinChannel "team" [bob] [alice, cath] shortLink fullLink frank
-          threadDelay 100000
+          threadDelay 1000000
           -- the rejected roster never elevates cath: the intro caps her to the channel default, so she
           -- stays observer (not moderator), and the version must not advance to the corrupted roster's version 1
           checkMemberRow frank "cath" (Just "observer")
@@ -10640,6 +10641,42 @@ testChannelPromotedMemberRejoinViaRelay ps =
           threadDelay 100000
           checkMemberRow dan "cath" (Just "member")
 
+testChannelRosterMultiRelayMultipart :: HasCallStack => TestParams -> IO ()
+testChannelRosterMultiRelayMultipart ps =
+  withNewTestChatCfgOpts ps cfg testOpts "alice" aliceProfile $ \alice ->
+    withNewTestChatCfgOpts ps cfg relayTestOpts "bob" bobProfile $ \bob ->
+      withNewTestChatCfgOpts ps cfg relayTestOpts "cath" cathProfile $ \cath ->
+        withNewTestChatCfgOpts ps cfg testOpts "dan" danProfile $ \dan ->
+          withNewTestChatCfgOpts ps cfg testOpts "eve" eveProfile $ \eve ->
+            withNewTestChatCfgOpts ps cfg testOpts "frank" frankProfile $ \frank -> do
+              createChannel2Relays "team" alice bob cath dan eve frank
+
+              -- promote eve to moderator: the owner-signed roster broadcasts through BOTH relays to dan and
+              -- frank (each connected to both). At fileChunkSize=30 the blob spans multiple chunks, so each
+              -- member receives two interleaved multi-chunk streams (one per relay) for the same roster.
+              threadDelay 1000000
+              alice ##> "/mr #team eve moderator"
+              alice <## "#team: you changed the role of eve to moderator (signed)"
+              concurrentlyN_
+                [ bob <## "#team: alice changed the role of eve from observer to moderator (signed)",
+                  cath <## "#team: alice changed the role of eve from observer to moderator (signed)",
+                  eve <## "#team: alice changed your role from observer to moderator (signed)",
+                  dan <### [EndsWith "to moderator (signed)"],
+                  frank <### [EndsWith "to moderator (signed)"]
+                ]
+              threadDelay 1000000 -- let both relays' interleaved multipart streams settle
+
+              -- per-source transfers keep the streams independent, so each member reassembles the blob and pins
+              -- eve as the single moderator WITH her owner-attested key (role + key both come from the blob)
+              checkOneModeratorWithKey dan
+              checkOneModeratorWithKey frank
+  where
+    cfg = testCfg {fileChunkSize = 30}
+    checkOneModeratorWithKey cc = do
+      rows <- withCCTransaction cc $ \db ->
+        DB.query_ db "SELECT member_pub_key FROM group_members WHERE member_role = 'moderator'" :: IO [Only (Maybe ByteString)]
+      map (\(Only k) -> isJust k) rows `shouldBe` [True]
+
 testChannelRemoveRelay :: HasCallStack => TestParams -> IO ()
 testChannelRemoveRelay ps =
   withNewTestChat ps "alice" aliceProfile $ \alice ->