diff --git a/changelog.d/6-federation/WPB-28422 b/changelog.d/6-federation/WPB-28422 new file mode 100644 index 00000000000..7ca489ce3bf --- /dev/null +++ b/changelog.d/6-federation/WPB-28422 @@ -0,0 +1 @@ +Remove stale local memberships when a remote conversation is definitively reported as not found. diff --git a/integration/test/Test/Conversation.hs b/integration/test/Test/Conversation.hs index 681df7f59fa..e3a859df46f 100644 --- a/integration/test/Test/Conversation.hs +++ b/integration/test/Test/Conversation.hs @@ -582,6 +582,189 @@ testGetOneOnOneConvInStatusSentFromRemote domain = do resp <- getConversation d1User d2ConvId resp.status `shouldMatchInt` 200 +testReconcileStaleLocalMembershipsForDeletedRemoteConversation :: (HasCallStack) => App () +testReconcileStaleLocalMembershipsForDeletedRemoteConversation = do + owner <- randomUser OwnDomain def + alice <- randomUser OtherDomain def + charlie <- randomUser OtherDomain def + for_ [alice, charlie] $ connectTwoUsers owner + + conv <- registerMissingRemoteConversation owner [alice, charlie] + + eventually $ do + assertConversationMembership alice conv True + assertConversationMembership charlie conv True + + let isSystemDeleteFor conversation event = + fieldEquals event "payload.0.type" "conversation.system.delete" + &&~ isNotifConv conversation event + + -- A successful response from the owning backend that omits the conversation + -- proves that Alice's locally stored membership is stale. + withWebSockets [alice, charlie] $ \[wsAlice, wsCharlie] -> do + getConversation alice conv >>= assertLabel 404 "no-conversation" + void $ awaitMatch (isSystemDeleteFor conv) wsAlice + assertConversationMembership alice conv False + assertConversationMembership charlie conv True + + -- The bulk endpoint performs the same reconciliation for Charlie. + bindResponse (listConversations charlie [conv]) $ \resp -> do + resp.status `shouldMatchInt` 200 + resp.json %. "found" `shouldMatch` ([] :: [Value]) + resp.json %. "not_found" `shouldMatch` [conv] + resp.json %. "failed" `shouldMatch` ([] :: [Value]) + void $ awaitMatch (isSystemDeleteFor conv) wsCharlie + + assertConversationMembership alice conv False + assertConversationMembership charlie conv False + + -- Reconciliation is idempotent once the local membership has been removed. + getConversation alice conv >>= assertLabel 404 "no-conversation" + +-- | Fetching stale remote conversations from two different domains reconciles +-- both independently. A response for one domain must not remove memberships +-- belonging to another domain. +testReconcileStaleMembershipsMultipleDomains :: (HasCallStack) => App () +testReconcileStaleMembershipsMultipleDomains = do + resourcePool <- asks resourcePool + runCodensity (acquireResources 1 resourcePool) $ \[remoteBackend] -> + runCodensity (startDynamicBackend remoteBackend mempty) $ \_ -> do + alice <- randomUser OwnDomain def + ownerStatic <- randomUser OtherDomain def + ownerDynamic <- randomUser remoteBackend.berDomain def + connectTwoUsers ownerStatic alice + connectTwoUsers ownerDynamic alice + convStatic <- registerMissingRemoteConversation ownerStatic [alice] + convDynamic <- registerMissingRemoteConversation ownerDynamic [alice] + + eventually $ do + assertConversationMembership alice convStatic True + assertConversationMembership alice convDynamic True + + bindResponse (listConversations alice [convStatic, convDynamic]) $ \resp -> do + resp.status `shouldMatchInt` 200 + resp.json %. "found" `shouldMatch` ([] :: [Value]) + resp.json %. "failed" `shouldMatch` ([] :: [Value]) + notFound <- resp.json %. "not_found" & asList + for_ [convStatic, convDynamic] $ \conv -> + (notFound :: [Value]) `shouldContain` [conv] + + assertConversationMembership alice convStatic False + assertConversationMembership alice convDynamic False + +-- | Conversations the remote still returns are preserved; only omitted ones +-- are reconciled within the same request. +testReconcileOnlyMissingConversations :: (HasCallStack) => App () +testReconcileOnlyMissingConversations = do + alice <- randomUser OwnDomain def + owner <- randomUser OtherDomain def + connectTwoUsers owner alice + + alive <- + postConversation owner (defProteus {qualifiedUsers = [alice]}) + >>= getJSON 201 + aliveQid <- objQidObject alive + stale <- registerMissingRemoteConversation owner [alice] + + eventually $ do + assertConversationMembership alice aliveQid True + assertConversationMembership alice stale True + + bindResponse (listConversations alice [aliveQid, stale]) $ \resp -> do + resp.status `shouldMatchInt` 200 + resp.json %. "failed" `shouldMatch` ([] :: [Value]) + found <- resp.json %. "found" & asList + length (found :: [Value]) `shouldMatchInt` 1 + notFound <- resp.json %. "not_found" & asList + (notFound :: [Value]) `shouldContain` [stale] + + assertConversationMembership alice aliveQid True + assertConversationMembership alice stale False + +testPreserveRemoteMembershipOnFederationFailure :: (HasCallStack) => App () +testPreserveRemoteMembershipOnFederationFailure = do + resourcePool <- asks resourcePool + runCodensity (acquireResources 1 resourcePool) $ \[remoteBackend] -> do + (alice, convQid) <- runCodensity (startDynamicBackend remoteBackend mempty) $ \_ -> do + owner <- randomUser remoteBackend.berDomain def + alice <- randomUser OwnDomain def + connectTwoUsers owner alice + conv <- + postConversation owner (defProteus {qualifiedUsers = [alice]}) + >>= getJSON 201 + convQid <- objQidObject conv + eventually $ assertConversationMembership alice convQid True + pure (alice, convQid) + + bindResponse (listConversations alice [convQid]) $ \resp -> do + resp.status `shouldMatchInt` 200 + resp.json %. "failed" `shouldMatch` [convQid] + assertConversationMembership alice convQid True + +-- | This function only submits the on-conversation-created event +-- to the remote backend without actually having created a local conversation. +registerMissingRemoteConversation :: (HasCallStack) => Value -> [Value] -> App Value +registerMissingRemoteConversation owner members = do + originDomain <- objDomain owner + originUserId <- objId owner + convId <- randomId + targetDomain <- case members of + [] -> assertFailure "A remote conversation needs at least one local member" + firstMember : remainingMembers -> do + domain <- objDomain firstMember + for_ remainingMembers $ \remoteMember -> do + memberDomain <- objDomain remoteMember + memberDomain `shouldMatch` domain + pure domain + memberPayloads <- for members $ \remoteMember -> do + memberId <- objId remoteMember + memberQid <- objQidObject remoteMember + pure + $ object + [ "id" .= memberId, + "qualified_id" .= memberQid, + "status" .= (0 :: Int), + "conversation_role" .= ("wire_member" :: String) + ] + req <- + rawBaseRequest + originDomain + FederatorInternal + Unversioned + (joinHttpPath ["rpc", targetDomain, "galley", "on-conversation-created"]) + bindResponse + ( submit "POST" + $ req + & addHeader "Wire-Origin-Domain" originDomain + & addJSONObject + [ "time" .= ("2026-01-01T00:00:00.000Z" :: String), + "orig_user_id" .= originUserId, + "cnv_id" .= convId, + "cnv_type" .= (0 :: Int), + "cnv_access" .= ["invite" :: String, "code"], + "cnv_access_roles" .= ["team_member" :: String, "non_team_member"], + "cnv_name" .= Aeson.Null, + "non_creator_members" .= memberPayloads, + "message_timer" .= Aeson.Null, + "receipt_mode" .= Aeson.Null, + "protocol" .= object ["protocol" .= ("proteus" :: String)], + "group_conv_type" .= ("group_conversation" :: String), + "channel_add_permission" .= Aeson.Null, + "history" .= Aeson.Null + ] + ) + $ \resp -> resp.status `shouldMatchInt` 200 + pure $ object ["domain" .= originDomain, "id" .= convId] + +assertConversationMembership :: (HasCallStack) => Value -> Value -> Bool -> App () +assertConversationMembership user conv expected = + bindResponse (listConversationIds user def) $ \resp -> do + resp.status `shouldMatchInt` 200 + conversationIds <- resp.json %. "qualified_conversations" & asList + if expected + then conversationIds `shouldContain` [conv] + else conversationIds `shouldNotContain` [conv] + testAddingUserNonFullyConnectedFederation :: (HasCallStack) => StaticDomain -> App () testAddingUserNonFullyConnectedFederation domain = do let overrides = @@ -1029,8 +1212,13 @@ testOnUserDeletedConversations = do do -- Bob is not in the one-to-one conversation with Alice any more - conv <- getConversation alice ooConvId >>= getJSON 200 - shouldBeEmpty $ conv %. "members.others" + resp <- getConversation alice ooConvId + case resp.status of + 200 -> do + conv <- getJSON 200 resp + shouldBeEmpty $ conv %. "members.others" + 404 -> resp.json %. "label" `shouldMatch` ("no-conversation" :: String) + status -> assertFailure $ "Unexpected status while fetching one-to-one conversation: " <> show status do -- Bob is not in the main conversation any more mainConvAfter <- getConversation alice (mainConvBefore %. "qualified_id") >>= getJSON 200 diff --git a/libs/wire-subsystems/src/Wire/ConversationSubsystem/Query.hs b/libs/wire-subsystems/src/Wire/ConversationSubsystem/Query.hs index c181a9c0866..3f020ae6f39 100644 --- a/libs/wire-subsystems/src/Wire/ConversationSubsystem/Query.hs +++ b/libs/wire-subsystems/src/Wire/ConversationSubsystem/Query.hs @@ -84,6 +84,7 @@ import Wire.API.Conversation.Role import Wire.API.Conversation.Role qualified as Public import Wire.API.Error import Wire.API.Error.Galley +import Wire.API.Event.Conversation (SystemEvent (..), SystemEventData (EdSystemConvDelete)) import Wire.API.Federation.API import Wire.API.Federation.API.Galley import Wire.API.Federation.Client (FederatorClient, getNegotiatedVersion) @@ -106,12 +107,16 @@ import Wire.ConversationSubsystem.Fetch (getConversationIdsImpl) import Wire.ConversationSubsystem.MLS import Wire.ConversationSubsystem.MLS.Enabled (assertMLSEnabled, getMLSPrivateKeys, isMLSEnabled) import Wire.ConversationSubsystem.MLS.One2One (localMLSOne2OneConversation, remoteMLSOne2OneConversation) +import Wire.ConversationSubsystem.Notify qualified as Notify import Wire.ConversationSubsystem.One2One import Wire.ConversationSubsystem.Util import Wire.FeaturesConfigSubsystem import Wire.FederationAPIAccess qualified as E import Wire.HashPassword (HashPassword) +import Wire.NotificationSubsystem import Wire.RateLimit +import Wire.Sem.Now (Now) +import Wire.Sem.Now qualified as Now import Wire.Sem.Paging.Cassandra import Wire.StoredConversation import Wire.StoredConversation qualified as Data @@ -184,6 +189,8 @@ getConversation :: Member (Error FederationError) r, Member (E.FederationAPIAccess FederatorClient) r, Member P.TinyLog r, + Member Now r, + Member NotificationSubsystem r, Member TeamSubsystem r ) => Local UserId -> @@ -205,6 +212,8 @@ getOwnConversation :: Member (Error InternalError) r, Member (E.FederationAPIAccess FederatorClient) r, Member P.TinyLog r, + Member Now r, + Member NotificationSubsystem r, Member TeamSubsystem r ) => Local UserId -> @@ -222,7 +231,9 @@ getRemoteConversation :: Member (ErrorS ConvNotFound) r, Member (Error FederationError) r, Member TinyLog r, - Member (E.FederationAPIAccess FederatorClient) r + Member (E.FederationAPIAccess FederatorClient) r, + Member Now r, + Member NotificationSubsystem r ) => Local UserId -> Remote ConvId -> @@ -239,7 +250,9 @@ getRemoteConversations :: Member (Error FederationError) r, Member (ErrorS 'ConvNotFound) r, Member (E.FederationAPIAccess FederatorClient) r, - Member P.TinyLog r + Member P.TinyLog r, + Member Now r, + Member NotificationSubsystem r ) => Local UserId -> [Remote ConvId] -> @@ -308,7 +321,9 @@ partitionGetConversationFailures = bimap concat concat . partitionEithers . map getRemoteConversationsWithFailures :: ( Member ConversationStore.ConversationStore r, Member (E.FederationAPIAccess FederatorClient) r, - Member P.TinyLog r + Member P.TinyLog r, + Member Now r, + Member NotificationSubsystem r ) => Local UserId -> [Remote ConvId] -> @@ -346,18 +361,38 @@ getRemoteConversationsWithFailures lusr convs = do rpc $ GetConversationsRequest (tUnqualified lusr) (tUnqualified someConvs) bimap (localFailures <>) (map remoteView . concat) . partitionEithers - <$> traverse handleFailure resp + <$> traverse (handleRequest locallyFound) resp where - handleFailure :: - (Member P.TinyLog r) => + handleRequest :: + ( Member ConversationStore.ConversationStore r, + Member P.TinyLog r, + Member Now r, + Member NotificationSubsystem r + ) => + [Remote ConvId] -> Either (Remote [ConvId], FederationError) (Remote GetRemoteConversationViewsResponse) -> Sem r (Either FailedGetConversation [Remote RemoteConversationView]) - handleFailure (Left (rcids, e)) = do + handleRequest _ (Left (rcids, e)) = do P.warn $ Logger.msg ("Error occurred while fetching remote conversations" :: ByteString) . Logger.field "error" (displayException e) pure . Left $ failedGetConversationRemotely (sequenceA rcids) e - handleFailure (Right c) = pure . Right . traverse (.convs) $ c + handleRequest locallyFound (Right response) = do + let locallyFoundForDomain = Set.fromList $ filter ((== tDomain response) . tDomain) locallyFound + returnedIds = Set.fromList $ map (qualifyAs response . (.id)) (tUnqualified response).convs + missingConversations = Set.toList $ locallyFoundForDomain `Set.difference` returnedIds + unless (null missingConversations) $ do + now <- Now.get + for_ missingConversations $ \conv -> do + ConversationStore.deleteMembersInRemoteConversation conv [tUnqualified lusr] + Notify.pushSystemEvent + Nothing + (SystemEvent (tUntagged conv) Nothing now Nothing EdSystemConvDelete) + (Set.singleton $ tUnqualified lusr) + P.info $ + Logger.msg ("Removed stale local memberships for remote conversations" :: ByteString) + . Logger.field "convIds" (show $ map tUntagged missingConversations) + pure . Right . traverse (.convs) $ response getConversationRoles :: ( Member ConversationStore.ConversationStore r, @@ -505,7 +540,9 @@ listConversations :: ( Member ConversationStore.ConversationStore r, Member (Error InternalError) r, Member (E.FederationAPIAccess FederatorClient) r, - Member P.TinyLog r + Member P.TinyLog r, + Member Now r, + Member NotificationSubsystem r ) => Local UserId -> Public.ListConversations -> @@ -529,9 +566,6 @@ listConversations luser (Public.ListConversations ids) = do fetchedOrFailedRemoteIds = Set.fromList $ map Public.cnvQualifiedId remoteConversations <> failedConvs remoteNotFoundRemoteIds = filter (`Set.notMember` fetchedOrFailedRemoteIds) $ map tUntagged remoteIds unless (null remoteNotFoundRemoteIds) $ - -- FUTUREWORK: This implies that the backends are out of sync. Maybe the - -- current user should be considered removed from this conversation at this - -- point. P.warn $ Logger.msg ("Some locally found conversation ids were not returned by remotes" :: ByteString) . Logger.field "convIds" (show remoteNotFoundRemoteIds) @@ -591,7 +625,9 @@ getSelfMember :: Member (ErrorS ConvNotFound) r, Member (Error FederationError) r, Member TinyLog r, - Member (E.FederationAPIAccess FederatorClient) r + Member (E.FederationAPIAccess FederatorClient) r, + Member Now r, + Member NotificationSubsystem r ) => Local UserId -> Qualified ConvId ->