diff --git a/src/Simplex/Chat/Library/Subscriber.hs b/src/Simplex/Chat/Library/Subscriber.hs index adc9bac759..fbce14124d 100644 --- a/src/Simplex/Chat/Library/Subscriber.hs +++ b/src/Simplex/Chat/Library/Subscriber.hs @@ -2913,9 +2913,9 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = COMContact c1@Contact {profile = p1} -> case cgm2 of COMGroupMember m2@GroupMember {memberProfile = p2, memberContactId} - | isNothing memberContactId && profilesMatch p1 p2 -> do - void . sendDirectContactMessage user c1 $ XInfoProbeOk probe - COMContact <$$> associateMemberAndContact c1 m2 + -- send XInfoProbeOk only if this side actually performed the association (won the concurrent race), otherwise stay silent + | isNothing memberContactId && profilesMatch p1 p2 -> + associateMemberAndContact c1 m2 >>= mapM (\c1' -> COMContact c1' <$ sendDirectContactMessage user c1 (XInfoProbeOk probe)) | otherwise -> messageWarning "probeMatch ignored: profiles don't match or member already has contact" >> pure Nothing COMContact _ -> messageWarning "probeMatch ignored: contacts are not merged" >> pure Nothing COMGroupMember m1@GroupMember {groupId, memberProfile = p1, memberContactId} -> @@ -2923,9 +2923,8 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = COMContact c2@Contact {profile = p2} | memberCurrent m1 && isNothing memberContactId && profilesMatch p1 p2 -> case memberConn m1 of - Just conn -> do - void $ sendDirectMemberMessage conn (XInfoProbeOk probe) groupId - COMContact <$$> associateMemberAndContact c2 m1 + Just conn -> + associateMemberAndContact c2 m1 >>= mapM (\c2' -> COMContact c2' <$ sendDirectMemberMessage conn (XInfoProbeOk probe) groupId) _ -> messageWarning "probeMatch ignored: matched member doesn't have connection" >> pure Nothing | otherwise -> messageWarning "probeMatch ignored: profiles don't match or member already has contact or member not current" >> pure Nothing COMGroupMember _ -> messageWarning "probeMatch ignored: members are not matched with members" >> pure Nothing @@ -3065,8 +3064,8 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = GroupMember {localDisplayName = mLDN} = m case (suffixOrd displayName cLDN, suffixOrd displayName mLDN) of (Just cOrd, Just mOrd) - | cOrd < mOrd -> Just <$> associateMemberWithContact c m - | mOrd < cOrd -> Just <$> associateContactWithMember m c + | cOrd < mOrd -> associateMemberWithContact c m + | mOrd < cOrd -> associateContactWithMember m c | otherwise -> pure Nothing _ -> pure Nothing @@ -3077,20 +3076,21 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = Just suffix -> readMaybe $ T.unpack suffix Nothing -> Nothing - associateMemberWithContact :: Contact -> GroupMember -> CM Contact + associateMemberWithContact :: Contact -> GroupMember -> CM (Maybe Contact) associateMemberWithContact c1 m2@GroupMember {groupId} = do - g <- withStore $ \db -> do - liftIO $ associateMemberWithContactRecord db user c1 m2 - getGroupInfo db cxt user groupId - toView $ CEvtContactAndMemberAssociated user c1 g m2 c1 - pure c1 + (won, g) <- withStore $ \db -> do + won <- liftIO $ associateMemberWithContactRecord db user c1 m2 + g <- getGroupInfo db cxt user groupId + pure (won, g) + when won $ toView $ CEvtContactAndMemberAssociated user c1 g m2 c1 + pure $ if won then Just c1 else Nothing - associateContactWithMember :: GroupMember -> Contact -> CM Contact + associateContactWithMember :: GroupMember -> Contact -> CM (Maybe Contact) associateContactWithMember m1@GroupMember {groupId} c2 = do - (c2', g) <- withStore $ \db -> + (c2'_, g) <- withStore $ \db -> liftM2 (,) (associateContactWithMemberRecord db cxt user m1 c2) (getGroupInfo db cxt user groupId) - toView $ CEvtContactAndMemberAssociated user c2 g m1 c2' - pure c2' + forM_ c2'_ $ \c2' -> toView $ CEvtContactAndMemberAssociated user c2 g m1 c2' + pure c2'_ saveConnInfo :: Connection -> ConnInfo -> CM (Connection, Maybe GroupInfo) saveConnInfo activeConn connInfo = do diff --git a/src/Simplex/Chat/Store/Groups.hs b/src/Simplex/Chat/Store/Groups.hs index b10b7ad239..41b2ff5d92 100644 --- a/src/Simplex/Chat/Store/Groups.hs +++ b/src/Simplex/Chat/Store/Groups.hs @@ -3067,52 +3067,56 @@ getContactOrMember_ db cxt user ids = (_, Just gId, Just gmId) -> COMGroupMember <$> getGroupMember db cxt user gId gmId _ -> throwError $ SEInternalError "" -associateMemberWithContactRecord :: DB.Connection -> User -> Contact -> GroupMember -> IO () +-- returns True if this call performed the association; concurrent workers racing to associate the same member all call this, only the first wins +associateMemberWithContactRecord :: DB.Connection -> User -> Contact -> GroupMember -> IO Bool associateMemberWithContactRecord db User {userId} Contact {contactId, localDisplayName, profile = LocalProfile {profileId}} GroupMember {groupId, groupMemberId, localDisplayName = memLDN, memberProfile = LocalProfile {profileId = memProfileId}} = do currentTs <- getCurrentTime - DB.execute - db - [sql| + -- atomic compare-and-set: WHERE contact_id IS NULL makes a losing concurrent association a no-op; RETURNING reports whether we won + won <- isJust <$> maybeFirstRow fromOnly (DB.query db [sql| UPDATE group_members SET contact_id = ?, local_display_name = ?, contact_profile_id = ?, updated_at = ? - WHERE user_id = ? AND group_id = ? AND group_member_id = ? - |] - (contactId, localDisplayName, profileId, currentTs, userId, groupId, groupMemberId) - when (memProfileId /= profileId) $ deleteUnusedProfile_ db userId memProfileId - when (memLDN /= localDisplayName) $ deleteUnusedDisplayName_ db userId memLDN + WHERE user_id = ? AND group_id = ? AND group_member_id = ? AND contact_id IS NULL + RETURNING 1 + |] (contactId, localDisplayName, profileId, currentTs, userId, groupId, groupMemberId) :: IO [Only Int]) + when won $ do + when (memProfileId /= profileId) $ deleteUnusedProfile_ db userId memProfileId + when (memLDN /= localDisplayName) $ deleteUnusedDisplayName_ db userId memLDN + pure won -associateContactWithMemberRecord :: DB.Connection -> StoreCxt -> User -> GroupMember -> Contact -> ExceptT StoreError IO Contact +-- returns Just the updated contact if this call performed the association, Nothing if a concurrent worker won the race first +associateContactWithMemberRecord :: DB.Connection -> StoreCxt -> User -> GroupMember -> Contact -> ExceptT StoreError IO (Maybe Contact) associateContactWithMemberRecord db cxt user@User {userId} GroupMember {groupId, groupMemberId, localDisplayName = memLDN, memberProfile = LocalProfile {profileId = memProfileId}} Contact {contactId, localDisplayName, profile = LocalProfile {profileId}} = do - liftIO $ do + won <- liftIO $ do currentTs <- getCurrentTime - DB.execute - db - [sql| + -- atomic compare-and-set: WHERE contact_id IS NULL makes a losing concurrent association a no-op; RETURNING reports whether we won + won <- isJust <$> maybeFirstRow fromOnly (DB.query db [sql| UPDATE group_members SET contact_id = ?, updated_at = ? - WHERE user_id = ? AND group_id = ? AND group_member_id = ? - |] - (contactId, currentTs, userId, groupId, groupMemberId) - DB.execute - db - [sql| - UPDATE contacts - SET local_display_name = ?, contact_profile_id = ?, updated_at = ? - WHERE user_id = ? AND contact_id = ? - |] - (memLDN, memProfileId, currentTs, userId, contactId) - when (profileId /= memProfileId) $ deleteUnusedProfile_ db userId profileId - when (localDisplayName /= memLDN) $ deleteUnusedDisplayName_ db userId localDisplayName - getContact db cxt user contactId + WHERE user_id = ? AND group_id = ? AND group_member_id = ? AND contact_id IS NULL + RETURNING 1 + |] (contactId, currentTs, userId, groupId, groupMemberId) :: IO [Only Int]) + when won $ do + DB.execute + db + [sql| + UPDATE contacts + SET local_display_name = ?, contact_profile_id = ?, updated_at = ? + WHERE user_id = ? AND contact_id = ? + |] + (memLDN, memProfileId, currentTs, userId, contactId) + when (profileId /= memProfileId) $ deleteUnusedProfile_ db userId profileId + when (localDisplayName /= memLDN) $ deleteUnusedDisplayName_ db userId localDisplayName + pure won + if won then Just <$> getContact db cxt user contactId else pure Nothing deleteUnusedDisplayName_ :: DB.Connection -> UserId -> ContactName -> IO () deleteUnusedDisplayName_ db userId localDisplayName = diff --git a/tests/ChatTests/Groups.hs b/tests/ChatTests/Groups.hs index 39146b93ad..93c5455cb8 100644 --- a/tests/ChatTests/Groups.hs +++ b/tests/ChatTests/Groups.hs @@ -137,7 +137,7 @@ chatGroupTests = do #if !defined(dbPostgres) -- TODO [postgres] restore from outdated db backup (same as in agent) describe "group message errors" $ do - it "show message decryption error" testGroupMsgDecryptError + it "show message decryption error" (withTestOutput testGroupMsgDecryptError) it "should report ratchet de-synchronization, synchronize ratchets" testGroupSyncRatchet it "synchronize ratchets, reset connection code" testGroupSyncRatchetCodeReset #endif @@ -3830,18 +3830,23 @@ setupDesynchronizedRatchet ps alice = do copyDb "bob" "bob_old" withTestChat ps "bob" $ \bob -> do bob <## "subscribed 2 connections on server localhost" + threadDelay 2000000 alice #> "#team 1" bob <# "#team alice> 1" bob #> "#team 2" alice <# "#team bob> 2" - alice #> "#team 3" - bob <# "#team alice> 3" - bob #> "#team 4" - alice <# "#team bob> 4" - withTestChat ps "bob_old" $ \bob -> do - bob <## "subscribed 2 connections on server localhost" + -- exchange ends with bob receiving so alice's delivery receipts for bob's messages are drained + -- from bob's queue (in-order) before close, otherwise a trailing receipt lingers and is + -- redelivered to the bob_old copy below, inflating its decryption-error count + bob #> "#team 3" + alice <# "#team bob> 3" + alice #> "#team 4" + bob <# "#team alice> 4" + -- /sync is prohibited while the ratchet is in sync; checked here on a deterministically in-sync connection (bob_old below may already detect desync on subscribe, racing the command) bob ##> "/sync #team alice" bob <## "error: command is prohibited, synchronizeRatchet: not allowed" + withTestChat ps "bob_old" $ \bob -> do + bob <## "subscribed 2 connections on server localhost" alice #> "#team 1" bob <## "#team alice: decryption error (connection out of sync), synchronization required" bob <## "use /sync #team alice to synchronize" @@ -6197,13 +6202,13 @@ testGroupHistoryDisappearingMessage = threadDelay 1000000 - -- 3 seconds so that messages 2 and 3 are not deleted for alice before sending history to cath - alice ##> "/set disappear #team on 4" + -- TTL is generous so messages 2 and 3 survive until cath receives them in history (adding a member is a multi-connection flow that can take a few seconds); deletions are awaited explicitly below + alice ##> "/set disappear #team on 8" alice <## "updated group preferences:" - alice <## "Disappearing messages: on (4 sec)" + alice <## "Disappearing messages: on (8 sec)" bob <## "alice updated group #team:" bob <## "updated group preferences:" - bob <## "Disappearing messages: on (4 sec)" + bob <## "Disappearing messages: on (8 sec)" bob #> "#team 2" alice <# "#team bob> 2" @@ -6247,6 +6252,9 @@ testGroupHistoryDisappearingMessage = r1 <- chat <$> getTermLine cath r1 `shouldContain` [(0, "1"), (0, "2"), (0, "3"), (0, "4")] + -- wait past the 8s TTL (counted from receipt, latest for cath who got them in history) so all deletions have fired and are buffered before we read them + threadDelay 9000000 + concurrentlyN_ [ alice <### [ "timed message deleted: 2",