diff --git a/changelog.d/5-internal/WPB-22954 b/changelog.d/5-internal/WPB-22954 new file mode 100644 index 00000000000..ecc0777a956 --- /dev/null +++ b/changelog.d/5-internal/WPB-22954 @@ -0,0 +1,3 @@ +Move mls commit locks from cassandra to postgresql advisory locks + +Locks are now pg advisory locks held on a dedicated pooled connection for the duration of a commit; contention answers stale-message immediately. No backfill worker or storage flag is needed; the cassandra mls_commit_locks table becomes unread and can be dropped in a follow-up. During a rolling restart of galley there is a brief mixed-arbiter window (old pods lock via cassandra, new pods via postgres); avoid running commits against one group across old and new pods simultaneously, and avoid rolling back after cutover. diff --git a/integration/integration.cabal b/integration/integration.cabal index 8dd9466197c..2195b8d0433 100644 --- a/integration/integration.cabal +++ b/integration/integration.cabal @@ -185,6 +185,7 @@ library Test.Migration.Util Test.MLS Test.MLS.Clients + Test.MLS.CommitLock Test.MLS.History Test.MLS.KeyPackage Test.MLS.Keys diff --git a/integration/test/Test/MLS/CommitLock.hs b/integration/test/Test/MLS/CommitLock.hs new file mode 100644 index 00000000000..f6ef6938d3d --- /dev/null +++ b/integration/test/Test/MLS/CommitLock.hs @@ -0,0 +1,38 @@ +-- This file is part of the Wire Server implementation. +-- +-- Copyright (C) 2026 Wire Swiss GmbH +-- +-- This program is free software: you can redistribute it and/or modify it under +-- the terms of the GNU Affero General Public License as published by the Free +-- Software Foundation, either version 3 of the License, or (at your option) any +-- later version. +-- +-- This program is distributed in the hope that it will be useful, but WITHOUT +-- ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS +-- FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more +-- details. +-- +-- You should have received a copy of the GNU Affero General Public License along +-- with this program. If not, see . + +module Test.MLS.CommitLock where + +import MLS.Util +import SetupHelpers +import Testlib.Prelude + +-- | Every MLS commit acquires and releases the commit lock, so two successive +-- commits prove acquire -> release -> re-acquire through the pg advisory-lock +-- interpreter. A leaked lock would fail the second commit. +testMLSCommitLock :: (HasCallStack) => App () +testMLSCommitLock = do + [alice, bob, charlie] <- createAndConnectUsers [OwnDomain, OwnDomain, OwnDomain] + alice1 <- createMLSClient def alice + bob1 <- createMLSClient def bob + void $ uploadNewKeyPackage def bob1 + convId <- createNewGroup def alice1 + void $ createAddCommit alice1 convId [bob] >>= sendAndConsumeCommitBundle + + charlie1 <- createMLSClient def charlie + void $ uploadNewKeyPackage def charlie1 + void $ createAddCommit alice1 convId [charlie] >>= sendAndConsumeCommitBundle diff --git a/libs/wire-subsystems/src/Wire/ConversationStore.hs b/libs/wire-subsystems/src/Wire/ConversationStore.hs index a8715073b8f..de97b4fe10a 100644 --- a/libs/wire-subsystems/src/Wire/ConversationStore.hs +++ b/libs/wire-subsystems/src/Wire/ConversationStore.hs @@ -23,7 +23,6 @@ import Data.Id import Data.Misc import Data.Qualified import Data.Range -import Data.Time.Clock import Imports import Polysemy import Wire.API.Conversation hiding (Conversation, Member) @@ -44,14 +43,11 @@ import Wire.Sem.Paging.Cassandra import Wire.StoredConversation import Wire.UserList -data LockAcquired - = Acquired - | NotAcquired - deriving (Show, Eq) - data MLSCommitLockStore m a where - AcquireCommitLock :: GroupId -> Epoch -> NominalDiffTime -> MLSCommitLockStore m LockAcquired - ReleaseCommitLock :: GroupId -> Epoch -> MLSCommitLockStore m () + -- | Runs the action while holding an exclusive lock for @(groupId, epoch)@. + -- Returns 'Nothing' without running the action when another holder is active + -- (callers respond 'MLSStaleMessage'). + HoldCommitLock :: GroupId -> Epoch -> m a -> MLSCommitLockStore m (Maybe a) data ConversationSearch = ConversationSearch { team :: TeamId, diff --git a/libs/wire-subsystems/src/Wire/ConversationStore/Cassandra.hs b/libs/wire-subsystems/src/Wire/ConversationStore/Cassandra.hs index 5f7fe31b5e7..dd6514412a0 100644 --- a/libs/wire-subsystems/src/Wire/ConversationStore/Cassandra.hs +++ b/libs/wire-subsystems/src/Wire/ConversationStore/Cassandra.hs @@ -16,8 +16,7 @@ -- with this program. If not, see . module Wire.ConversationStore.Cassandra - ( interpretMLSCommitLockStoreToCassandra, - interpretConversationStoreToCassandra, + ( interpretConversationStoreToCassandra, interpretConversationStoreToCassandraAndPostgres, interpretConversationStoreByMigration, MigrationError (..), @@ -41,7 +40,6 @@ import Data.Monoid import Data.Qualified import Data.Range import Data.Set qualified as Set -import Data.Time import Data.UUID.Util qualified as UUID import Imports import Network.HTTP.Types.Status (status500) @@ -71,7 +69,7 @@ import Wire.API.MLS.GroupInfo import Wire.API.MLS.LeafNode (LeafIndex) import Wire.API.MLS.SubConversation import Wire.API.Provider.Service -import Wire.ConversationStore (ConversationStore (..), LockAcquired (..), MLSCommitLockStore (..)) +import Wire.ConversationStore (ConversationStore (..)) import Wire.ConversationStore qualified as ConvStore import Wire.ConversationStore.Cassandra.Instances () import Wire.ConversationStore.Cassandra.Queries qualified as Cql @@ -351,37 +349,6 @@ updateToMLSProtocol client cnv = updateChannelAddPermissions :: ConvId -> AddPermission -> Client () updateChannelAddPermissions cid cap = retry x5 $ write Cql.updateChannelAddPermission (params LocalQuorum (cap, cid)) -acquireCommitLock :: GroupId -> Epoch -> NominalDiffTime -> Client LockAcquired -acquireCommitLock groupId epoch ttl = do - rows <- - retry x5 $ - trans - Cql.acquireCommitLock - ( params - LocalQuorum - (groupId, epoch, round ttl) - ) - { serialConsistency = Just LocalSerialConsistency - } - pure $ - if checkTransSuccess rows - then Acquired - else NotAcquired - -releaseCommitLock :: GroupId -> Epoch -> Client () -releaseCommitLock groupId epoch = - retry x5 $ - write - Cql.releaseCommitLock - ( params - LocalQuorum - (groupId, epoch) - ) - -checkTransSuccess :: [Row] -> Bool -checkTransSuccess [] = False -checkTransSuccess (row : _) = either (const False) (fromMaybe False) $ fromRow 0 row - removeTeamConv :: TeamId -> ConvId -> Client () removeTeamConv tid cid = liftClient $ do retry x5 . batch $ do @@ -880,15 +847,6 @@ isConversationOutOfSync cid = maybe False (fromMaybe False . runIdentity) <$> retry x1 (query1 Cql.lookupConvOutOfSync (params LocalQuorum (Identity cid))) -interpretMLSCommitLockStoreToCassandra :: (Member (Embed IO) r, Member TinyLog r) => ClientState -> InterpreterFor MLSCommitLockStore r -interpretMLSCommitLockStoreToCassandra client = interpret $ \case - AcquireCommitLock gId epoch ttl -> do - logEffect "MLSCommitLockStore.AcquireCommitLock" - embedClient client $ acquireCommitLock gId epoch ttl - ReleaseCommitLock gId epoch -> do - logEffect "MLSCommitLockStore.ReleaseCommitLock" - embedClient client $ releaseCommitLock gId epoch - interpretConversationStoreToCassandra :: forall r a. ( PGConstraints r, diff --git a/libs/wire-subsystems/src/Wire/ConversationStore/Cassandra/Queries.hs b/libs/wire-subsystems/src/Wire/ConversationStore/Cassandra/Queries.hs index fef06f288bb..a75a144bcf9 100644 --- a/libs/wire-subsystems/src/Wire/ConversationStore/Cassandra/Queries.hs +++ b/libs/wire-subsystems/src/Wire/ConversationStore/Cassandra/Queries.hs @@ -21,7 +21,6 @@ -- - conversation -- - member -- - member_remote_user --- - mls_commit_locks -- - mls_group_member_client -- - subconversation -- - team_conv @@ -355,12 +354,6 @@ removeAllMLSClients = "DELETE FROM mls_group_member_client WHERE group_id = ?" lookupMLSClients :: PrepQuery R (Identity GroupId) (Domain, UserId, ClientId, Int32, Bool) lookupMLSClients = "select user_domain, user, client, leaf_node_index, removal_pending from mls_group_member_client where group_id = ?" -acquireCommitLock :: PrepQuery W (GroupId, Epoch, Int32) Row -acquireCommitLock = "insert into mls_commit_locks (group_id, epoch) values (?, ?) if not exists using ttl ?" - -releaseCommitLock :: PrepQuery W (GroupId, Epoch) () -releaseCommitLock = "delete from mls_commit_locks where group_id = ? and epoch = ?" - -- Bots --------------------------------------------------------------------- insertBot :: PrepQuery W (ConvId, BotId, ServiceId, ProviderId) () diff --git a/libs/wire-subsystems/src/Wire/ConversationSubsystem/Action/Reset.hs b/libs/wire-subsystems/src/Wire/ConversationSubsystem/Action/Reset.hs index c31399a3b4d..732cc55c695 100644 --- a/libs/wire-subsystems/src/Wire/ConversationSubsystem/Action/Reset.hs +++ b/libs/wire-subsystems/src/Wire/ConversationSubsystem/Action/Reset.hs @@ -26,7 +26,6 @@ import Imports import Polysemy import Polysemy.Error import Polysemy.Input -import Polysemy.Resource import Polysemy.TinyLog qualified as P import System.Logger.Class qualified as Log import Wire.API.Conversation hiding (Member) @@ -67,7 +66,6 @@ resetLocalMLSMainConversation :: Member NotificationSubsystem r, Member ProposalStore r, Member Random r, - Member Resource r, Member ConversationStore r, Member P.TinyLog r, Member MLSCommitLockStore r, diff --git a/libs/wire-subsystems/src/Wire/ConversationSubsystem/Federation.hs b/libs/wire-subsystems/src/Wire/ConversationSubsystem/Federation.hs index 93e4aa44d97..20f9b14ced2 100644 --- a/libs/wire-subsystems/src/Wire/ConversationSubsystem/Federation.hs +++ b/libs/wire-subsystems/src/Wire/ConversationSubsystem/Federation.hs @@ -687,7 +687,6 @@ sendMLSCommitBundle :: Member (Input (Maybe (MLSKeysByPurpose MLSPrivateKeys))) r, Member Now r, Member LegalHoldStore r, - Member Resource r, Member TeamStore r, Member FederationSubsystem r, Member TeamSubsystem r, @@ -800,7 +799,6 @@ leaveSubConversation :: ( HasLeaveSubConversationEffects r, Member (Error FederationError) r, Member (Input (Local ())) r, - Member Resource r, Member TeamSubsystem r, Member E.MLSCommitLockStore r, Member (Input ConversationSubsystemConfig) r @@ -823,7 +821,6 @@ leaveSubConversation domain lscr = do deleteSubConversationForRemoteUser :: ( Member E.ConversationStore r, Member (Input (Local ())) r, - Member Resource r, Member TeamSubsystem r, Member E.MLSCommitLockStore r, Member TinyLog r diff --git a/libs/wire-subsystems/src/Wire/ConversationSubsystem/MLS/Commit/ExternalCommit.hs b/libs/wire-subsystems/src/Wire/ConversationSubsystem/MLS/Commit/ExternalCommit.hs index c711391146c..68342b5a26b 100644 --- a/libs/wire-subsystems/src/Wire/ConversationSubsystem/MLS/Commit/ExternalCommit.hs +++ b/libs/wire-subsystems/src/Wire/ConversationSubsystem/MLS/Commit/ExternalCommit.hs @@ -31,7 +31,6 @@ import Data.Set qualified as Set import Imports import Polysemy import Polysemy.Error -import Polysemy.Resource (Resource) import Polysemy.State import Wire.API.Conversation.Protocol import Wire.API.Error @@ -136,7 +135,6 @@ processExternalCommit :: Member (ErrorS MLSStaleMessage) r, Member (ErrorS MLSIdentityMismatch) r, Member (ErrorS MLSSubConvClientNotInParent) r, - Member Resource r, HasProposalActionEffects r, Member (ErrorS MLSInvalidLeafNodeSignature) r, Member MLSCommitLockStore r diff --git a/libs/wire-subsystems/src/Wire/ConversationSubsystem/MLS/Commit/InternalCommit.hs b/libs/wire-subsystems/src/Wire/ConversationSubsystem/MLS/Commit/InternalCommit.hs index 68bb12c1ec0..69f9de6bd7c 100644 --- a/libs/wire-subsystems/src/Wire/ConversationSubsystem/MLS/Commit/InternalCommit.hs +++ b/libs/wire-subsystems/src/Wire/ConversationSubsystem/MLS/Commit/InternalCommit.hs @@ -36,7 +36,6 @@ import Polysemy.Async (Async) import Polysemy.Async qualified as P import Polysemy.Error import Polysemy.Input (Input) -import Polysemy.Resource (Resource) import Wire.API.Conversation hiding (Member) import Wire.API.Conversation.Action import Wire.API.Conversation.Config (ConversationSubsystemConfig) @@ -79,7 +78,6 @@ processInternalCommit :: Member (ErrorS 'MLSIdentityMismatch) r, Member (ErrorS 'MissingLegalholdConsent) r, Member (ErrorS 'GroupIdVersionNotSupported) r, - Member Resource r, Member Async r, Member Random r, Member (ErrorS MLSInvalidLeafNodeSignature) r, diff --git a/libs/wire-subsystems/src/Wire/ConversationSubsystem/MLS/Message.hs b/libs/wire-subsystems/src/Wire/ConversationSubsystem/MLS/Message.hs index 760bcd63e22..5d69c11dc4f 100644 --- a/libs/wire-subsystems/src/Wire/ConversationSubsystem/MLS/Message.hs +++ b/libs/wire-subsystems/src/Wire/ConversationSubsystem/MLS/Message.hs @@ -47,7 +47,6 @@ import Polysemy.Async (Async) import Polysemy.Error import Polysemy.Input import Polysemy.Output -import Polysemy.Resource (Resource) import Polysemy.TinyLog import System.Logger qualified as Log import Wire.API.Conversation hiding (Member) @@ -176,7 +175,6 @@ postMLSCommitBundle :: Member (ErrorS GroupIdVersionNotSupported) r, Member (Input (Maybe GroupInfoCheckEnabled)) r, Member Random r, - Member Resource r, Members MLSMessageStaticErrors r, Member (ErrorS 'MLSInvalidLeafNodeSignature) r, HasProposalEffects r, @@ -214,7 +212,6 @@ postMLSCommitBundleFromLocalUser :: Member (Input (Maybe GroupInfoCheckEnabled)) r, Member (Input (Maybe (MLSKeysByPurpose MLSPrivateKeys))) r, Member Random r, - Member Resource r, Members MLSMessageStaticErrors r, Member (ErrorS 'MLSInvalidLeafNodeSignature) r, HasProposalEffects r, @@ -252,7 +249,6 @@ postMLSCommitBundleToLocalConv :: Member (Input EnableOutOfSyncCheck) r, Member (Input (Maybe GroupInfoCheckEnabled)) r, Member Random r, - Member Resource r, Members MLSMessageStaticErrors r, Member (ErrorS 'MLSInvalidLeafNodeSignature) r, HasProposalEffects r, diff --git a/libs/wire-subsystems/src/Wire/ConversationSubsystem/MLS/SubConversation.hs b/libs/wire-subsystems/src/Wire/ConversationSubsystem/MLS/SubConversation.hs index bcb660e72ec..092bcd20494 100644 --- a/libs/wire-subsystems/src/Wire/ConversationSubsystem/MLS/SubConversation.hs +++ b/libs/wire-subsystems/src/Wire/ConversationSubsystem/MLS/SubConversation.hs @@ -39,7 +39,6 @@ import Imports import Polysemy import Polysemy.Error import Polysemy.Input -import Polysemy.Resource import Polysemy.TinyLog import Wire.API.Conversation hiding (Member) import Wire.API.Conversation.Config (ConversationSubsystemConfig) @@ -218,7 +217,6 @@ deleteSubConversation :: Member (Error FederationError) r, Member (FederationAPIAccess FederatorClient) r, Member (Input (Maybe (MLSKeysByPurpose MLSPrivateKeys))) r, - Member Resource r, Member Conversation.MLSCommitLockStore r, Member TeamSubsystem r, Member TinyLog r @@ -292,7 +290,6 @@ leaveSubConversation :: Member (Error FederationError) r, Member (ErrorS 'MLSStaleMessage) r, Member (ErrorS 'MLSNotEnabled) r, - Member Resource r, Members LeaveSubConversationStaticErrors r, Member Conversation.MLSCommitLockStore r, Member TeamSubsystem r, @@ -319,7 +316,6 @@ leaveLocalSubConversation :: Member (ErrorS 'MLSStaleMessage) r, Member (ErrorS 'MLSNotEnabled) r, Member (Error FederationError) r, - Member Resource r, Members LeaveSubConversationStaticErrors r, Member Conversation.MLSCommitLockStore r, Member TeamSubsystem r, @@ -394,7 +390,6 @@ resetLocalSubConversation :: Member (ErrorS 'ConvAccessDenied) r, Member (ErrorS 'ConvNotFound) r, Member (ErrorS 'MLSStaleMessage) r, - Member Resource r, Member Conversation.MLSCommitLockStore r, Member TeamSubsystem r, Member TinyLog r diff --git a/libs/wire-subsystems/src/Wire/ConversationSubsystem/MLS/Util.hs b/libs/wire-subsystems/src/Wire/ConversationSubsystem/MLS/Util.hs index b862e488b09..56351a8e322 100644 --- a/libs/wire-subsystems/src/Wire/ConversationSubsystem/MLS/Util.hs +++ b/libs/wire-subsystems/src/Wire/ConversationSubsystem/MLS/Util.hs @@ -28,7 +28,6 @@ import Data.Text qualified as T import Imports import Polysemy import Polysemy.Error -import Polysemy.Resource (Resource, bracket) import Polysemy.TinyLog (TinyLog) import Polysemy.TinyLog qualified as TinyLog import System.Logger qualified as Log @@ -100,8 +99,7 @@ getPendingBackendRemoveProposals gid epoch = do withCommitLock :: forall r. - ( Member Resource r, - Member ConversationStore r, + ( Member ConversationStore r, Member (ErrorS 'MLSStaleMessage) r, Member MLSCommitLockStore r, Member TinyLog r @@ -110,37 +108,33 @@ withCommitLock :: GroupId -> Epoch -> Codensity (Sem r) () -withCommitLock lConvOrSubId gid epoch = - Codensity $ \k -> - bracket - ( acquireCommitLock gid epoch ttl >>= \lockAcquired -> - when (lockAcquired == NotAcquired) $ do - logStaleCommitLock - "commit-lock-not-acquired" - lConvOrSubId - gid - epoch - Nothing - throwS @'MLSStaleMessage - ) - (const $ releaseCommitLock gid epoch) - ( const $ do - actualEpoch <- - fromMaybe (Epoch 0) <$> case tUnqualified lConvOrSubId of - Conv cnv -> getConversationEpoch cnv - SubConv cnv sub -> getSubConversationEpoch cnv sub - unless (actualEpoch == epoch) $ do - logStaleCommitLock - "commit-lock-epoch-mismatch" - lConvOrSubId - gid - epoch - (Just actualEpoch) - throwS @'MLSStaleMessage - k () - ) - where - ttl = fromIntegral (600 :: Int) -- 10 minutes +withCommitLock lConvOrSubId gid epoch = Codensity $ \k -> do + committed <- holdCommitLock gid epoch $ do + actualEpoch <- + fromMaybe (Epoch 0) <$> case tUnqualified lConvOrSubId of + Conv cnv -> getConversationEpoch cnv + SubConv cnv sub -> getSubConversationEpoch cnv sub + unless (actualEpoch == epoch) $ do + logStaleCommitLock + "commit-lock-epoch-mismatch" + lConvOrSubId + gid + epoch + (Just actualEpoch) + throwS @'MLSStaleMessage + k () + maybe + ( do + logStaleCommitLock + "commit-lock-not-acquired" + lConvOrSubId + gid + epoch + Nothing + throwS @'MLSStaleMessage + ) + pure + committed logStaleCommitLock :: (Member TinyLog r) => diff --git a/libs/wire-subsystems/src/Wire/MLSCommitLockStore/Postgres.hs b/libs/wire-subsystems/src/Wire/MLSCommitLockStore/Postgres.hs new file mode 100644 index 00000000000..70512f5aff1 --- /dev/null +++ b/libs/wire-subsystems/src/Wire/MLSCommitLockStore/Postgres.hs @@ -0,0 +1,83 @@ +{-# OPTIONS_GHC -Wno-orphans #-} + +-- This file is part of the Wire Server implementation. +-- +-- Copyright (C) 2026 Wire Swiss GmbH +-- +-- This program is free software: you can redistribute it and/or modify it under +-- the terms of the GNU Affero General Public License as published by the Free +-- Software Foundation, either version 3 of the License, or (at your option) any +-- later version. +-- +-- This program is distributed in the hope that it will be useful, but WITHOUT +-- ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS +-- FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more +-- details. +-- +-- You should have received a copy of the GNU Affero General Public License along +-- with this program. If not, see . + +module Wire.MLSCommitLockStore.Postgres + ( interpretMLSCommitLockStoreToPostgres, + ) +where + +import Data.Bits (rotateL, xor) +import Data.Hashable (hash) +import Data.Hex (hex) +import Data.Text qualified as Text +import Data.Text.Encoding (decodeUtf8) +import Imports +import Polysemy +import Polysemy.Resource (Resource) +import Polysemy.TinyLog (TinyLog) +import Wire.API.MLS.Epoch +import Wire.API.MLS.Group +import Wire.ConversationStore (MLSCommitLockStore (..)) +import Wire.MigrationLock +import Wire.Postgres (PGConstraints) + +-- | Postgres interpreter for 'MLSCommitLockStore'. +-- +-- Implements the lock as a session-scoped pg advisory lock (via +-- 'tryWithMigrationLock') on a dedicated pooled connection, held for the +-- duration of the action and released on completion — or when the holding +-- connection dies, which replaces the Cassandra 10-minute TTL as the crash +-- guard. Contention returns 'Nothing' immediately so callers respond with +-- @stale-message@ without waiting. +-- +-- Accepted trade-offs (inherent to advisory locks): there is no TTL, so a +-- hung-but-alive holder blocks the group until its connection drops; locks +-- are not replicated and vanish on a Postgres failover; and the key is a +-- hashed @Int64@, so a hash collision yields a spurious stale response (the +-- same approach accepted for the existing @(TeamId, Text)@ instance). +interpretMLSCommitLockStoreToPostgres :: + ( PGConstraints r, + Member Resource r, + Member TinyLog r + ) => + InterpreterFor MLSCommitLockStore r +interpretMLSCommitLockStoreToPostgres = interpretH $ \case + HoldCommitLock gId epoch action -> do + m <- runT action + let run_it = raise . interpretMLSCommitLockStoreToPostgres + r <- run_it $ + tryWithMigrationLock (gId, epoch) $ do + fa <- m + pure (Just <$> fa) + case r of + Just x -> pure x + Nothing -> pureT Nothing + +-- | Combines group id and epoch into one lock key; rotate+xor mixes the two +-- hashes to reduce collisions. +instance MigrationLockable (GroupId, Epoch) where + lockScope = "mls_commit_lock" + lockKey (gId, epoch) = + (fromIntegral (hash (unGroupId gId)) :: Int64) `rotateL` 31 + `xor` fromIntegral (epochNumber epoch) + toText (gId, epoch) = + "0x" + <> decodeUtf8 (hex (unGroupId gId)) + <> ":" + <> Text.pack (show (epochNumber epoch)) diff --git a/libs/wire-subsystems/src/Wire/MigrationLock.hs b/libs/wire-subsystems/src/Wire/MigrationLock.hs index c75bec048cb..6b0869dd919 100644 --- a/libs/wire-subsystems/src/Wire/MigrationLock.hs +++ b/libs/wire-subsystems/src/Wire/MigrationLock.hs @@ -36,6 +36,8 @@ module Wire.MigrationLock where +import Control.Concurrent.Async qualified as Async +import Control.Exception (onException) import Data.Bits import Data.Hashable (hash) import Data.Id @@ -60,6 +62,7 @@ import Polysemy.Time.Data.TimeUnit import Polysemy.TinyLog (TinyLog) import Polysemy.TinyLog qualified as TinyLog import System.Logger.Message qualified as Log +import System.Timeout qualified import Wire.API.Error import Wire.API.PostgresMarshall import Wire.Error @@ -176,6 +179,121 @@ withMigrationLocks lockType maxWait lockables action = do FROM (SELECT pg_advisory_unlock_shared(lockId) FROM (SELECT UNNEST($1 :: bigint[]) as lockId) AS t) AS t2|] +-- | Non-blocking variant of 'withMigrationLocks' for a single lock: acquires a +-- session-scoped advisory lock for the key on a dedicated pooled connection, +-- runs the action, releases. Returns 'Nothing' without running the action if +-- the key is already locked. This is a try-lock (instead of the blocking +-- 'withMigrationLocks') because instant 'Nothing' preserves the existing +-- not-acquired -> stale-message client behavior. +-- +-- The release is bracketed with 'Polysemy.Resource.bracket', so it also runs +-- when the action short-circuits via other effects (e.g. an error response). +tryWithMigrationLock :: + forall x a r. + ( PGConstraints r, + Member Resource r, + Member TinyLog r, + MigrationLockable x + ) => + x -> + Sem r a -> + Sem r (Maybe a) +tryWithMigrationLock lockable action = + tryAcquireMigrationLock lockable >>= \case + Nothing -> pure Nothing + Just token -> + Just + <$> bracket + (pure ()) + (const (releaseMigrationLock lockable token)) + (const action) + +-- | Opaque handle to the connection holding an acquired advisory lock. +data MigrationLockToken = MigrationLockToken + { actionCompleted :: MVar (), + lockThread :: Async.Async () + } + +-- | Non-blocking acquire of a session-scoped advisory lock for the key on a +-- dedicated pooled connection. Returns 'Nothing' without side effects if the +-- key is already locked. +tryAcquireMigrationLock :: + forall x r. + ( PGConstraints r, + MigrationLockable x + ) => + x -> + Sem r (Maybe MigrationLockToken) +tryAcquireMigrationLock lockable = do + lockAcquired <- embed newEmptyMVar + actionCompleted <- embed newEmptyMVar + + pool <- (.rawPool) <$> input @HasqlPoolExt.Pool + lockThread <- + embed . Async.async $ + let holdSession = + Hasql.use pool $ do + ok <- Session.statement (lockKey lockable) tryAcquireLock + liftIO $ putMVar lockAcquired (Right ok) + when ok $ do + liftIO $ takeMVar actionCompleted + Session.statement [lockKey lockable] releaseLock + -- If the session failed before signaling (e.g. connection error), + -- the caller would otherwise block forever on 'lockAcquired'. + signalFailure = \case + Left e -> void (tryPutMVar lockAcquired (Left e)) + Right _ -> pure () + in holdSession >>= signalFailure + + -- Cancelling the thread ends its session, which releases any advisory lock + -- it may already have taken; without this an async exception while waiting + -- here would leak the lock. + acquired <- embed $ takeMVar lockAcquired `onException` Async.cancel lockThread + case acquired of + Left e -> do + embed $ Async.cancel lockThread + throw e + Right False -> do + embed $ Async.cancel lockThread + pure Nothing + Right True -> + pure . Just $ MigrationLockToken {actionCompleted, lockThread} + +-- | Release a lock acquired with 'tryAcquireMigrationLock'. Signals the +-- holding connection to unlock and gives it ~1s to finish cleanly. +releaseMigrationLock :: + forall x r. + ( PGConstraints r, + Member TinyLog r, + MigrationLockable x + ) => + x -> + MigrationLockToken -> + Sem r () +releaseMigrationLock lockable token = do + let MigrationLockToken {actionCompleted, lockThread} = token + logError errorStr = + TinyLog.warn $ + Log.msg (Log.val "Failed to cleanly unlock the migration locks") + . Log.field ("scope_" <> lockScope @x) (lockKey lockable) + . Log.field "error" errorStr + _ <- embed $ tryPutMVar actionCompleted () + mRes <- embed $ System.Timeout.timeout 1_000_000 (Async.wait lockThread) + case mRes of + Nothing -> logError ("timed out waiting for unlock" :: Text) + Just () -> pure () + +tryAcquireLock :: Hasql.Statement Int64 Bool +tryAcquireLock = + [singletonStatement|SELECT (pg_try_advisory_lock($1 :: bigint) :: bool)|] + +releaseLock :: Hasql.Statement [Int64] () +releaseLock = + lmapPG @(Vector _) + [resultlessStatement|SELECT (1 :: int) + FROM (SELECT pg_advisory_unlock(lockId) + FROM (SELECT UNNEST($1 :: bigint[]) as lockId))|] + -------------------------------------------------------------------------------- -- Combines team id and feature name into one lock key to keep per-feature locks distinct within a team -- without introducing a separate lock table; rotate+xor mixes the two hashes to reduce collisions. diff --git a/libs/wire-subsystems/wire-subsystems.cabal b/libs/wire-subsystems/wire-subsystems.cabal index 66662926e3d..efb31a6feff 100644 --- a/libs/wire-subsystems/wire-subsystems.cabal +++ b/libs/wire-subsystems/wire-subsystems.cabal @@ -404,6 +404,7 @@ library Wire.MeetingsSubsystem.Notification Wire.Migration Wire.MigrationLock + Wire.MLSCommitLockStore.Postgres Wire.MlsKeyPackageStore Wire.MlsKeyPackageStore.Cassandra Wire.MlsKeyPackageSubsystem diff --git a/services/background-worker/src/Wire/Effects.hs b/services/background-worker/src/Wire/Effects.hs index 0d367f19eed..6988c406629 100644 --- a/services/background-worker/src/Wire/Effects.hs +++ b/services/background-worker/src/Wire/Effects.hs @@ -73,7 +73,7 @@ import Wire.CodeStore.Cassandra (interpretCodeStoreToCassandra) import Wire.CodeStore.DualWrite (interpretCodeStoreToCassandraAndPostgres) import Wire.CodeStore.Postgres (interpretCodeStoreToPostgres) import Wire.ConversationStore (ConversationStore, MLSCommitLockStore) -import Wire.ConversationStore.Cassandra (MigrationError (..), interpretConversationStoreByMigration, interpretMLSCommitLockStoreToCassandra) +import Wire.ConversationStore.Cassandra (MigrationError (..), interpretConversationStoreByMigration) import Wire.ConversationSubsystem (ConversationSubsystem) import Wire.ConversationSubsystem.Interpreter (ConversationSubsystemError, GroupInfoCheckEnabled (..), IntraListing (..), interpretConversationSubsystem) import Wire.ExternalAccess (ExternalAccess) @@ -96,6 +96,7 @@ import Wire.JobSubsystem.Interpreter (interpretJobSubsystem) import Wire.LegalHoldStore (LegalHoldStore) import Wire.LegalHoldStore.Cassandra (interpretLegalHoldStoreToCassandra) import Wire.LegalHoldStore.Env (LegalHoldEnv (..)) +import Wire.MLSCommitLockStore.Postgres (interpretMLSCommitLockStoreToPostgres) import Wire.MeetingNotifier (MeetingNotifier) import Wire.MeetingNotifier.NoOpInterpreter (discardMeetingNotifier) import Wire.MigrationLock (MigrationLockError) @@ -326,7 +327,7 @@ runBackgroundWorkerEffects env extEnv requestId mJobId = . runInputConst @(Maybe GroupInfoCheckEnabled) (GroupInfoCheckEnabled <$> env.checkGroupInfo) . runInputConst @(Maybe GuestLinkTTLSeconds) env.guestLinkTTLSeconds . runInputConst @FanoutLimit (currentFanoutLimit env.maxTeamSize env.maxFanoutSize) - . interpretMLSCommitLockStoreToCassandra env.cassandraGalley + . interpretMLSCommitLockStoreToPostgres . interpretProposalStoreToCassandra . interpretServiceStoreToCassandra env.cassandraBrig . interpretUserGroupStoreToPostgres diff --git a/services/galley/src/Galley/App.hs b/services/galley/src/Galley/App.hs index 4d755d8c612..62e99f4a40b 100644 --- a/services/galley/src/Galley/App.hs +++ b/services/galley/src/Galley/App.hs @@ -108,7 +108,7 @@ import Wire.CodeStore.Cassandra import Wire.CodeStore.DualWrite import Wire.CodeStore.Postgres import Wire.ConversationStore (ConversationStore, MLSCommitLockStore) -import Wire.ConversationStore.Cassandra (MigrationError (..), interpretConversationStoreByMigration, interpretMLSCommitLockStoreToCassandra) +import Wire.ConversationStore.Cassandra (MigrationError (..), interpretConversationStoreByMigration) import Wire.ConversationSubsystem import Wire.ConversationSubsystem.Interpreter (ConversationSubsystemError, GroupInfoCheckEnabled (..), IntraListing (IntraListing), interpretConversationSubsystem) import Wire.CustomBackendStore @@ -137,6 +137,7 @@ import Wire.ListItems.Team.Cassandra ( interpretInternalTeamListToCassandra, interpretTeamListToCassandra, ) +import Wire.MLSCommitLockStore.Postgres (interpretMLSCommitLockStoreToPostgres) import Wire.MeetingNotifier (MeetingNotifier) import Wire.MeetingNotifier.Interpreter (interpretMeetingNotifier) import Wire.MeetingsStore (MeetingsStore) @@ -542,7 +543,7 @@ evalGalley e = . interpretTeamMemberStoreToCassandraWithPaging lh . interpretTeamMemberStoreToCassandra lh . teamFeatureStoreInterpreter - . interpretMLSCommitLockStoreToCassandra (e ^. cstate) + . interpretMLSCommitLockStoreToPostgres . convStoreInterpreter . interpretTeamNotificationStoreToCassandra . interpretServiceStoreToCassandra (e ^. cstate) diff --git a/services/galley/test/integration/API/MLS.hs b/services/galley/test/integration/API/MLS.hs index ab2234af0ea..2249e449f61 100644 --- a/services/galley/test/integration/API/MLS.hs +++ b/services/galley/test/integration/API/MLS.hs @@ -24,7 +24,6 @@ import API.MLS.Util import API.Util import Bilge hiding (empty, head) import Bilge.Assert -import Cassandra hiding (Set) import Control.Lens (view) import Control.Lens.Extras import Control.Monad.State qualified as State @@ -94,7 +93,6 @@ tests s = test s "add client of existing user" testAddClientPartial, test s "add user with some non-MLS clients" testAddUserWithProteusClients, test s "add remote users to a conversation (some unreachable)" testAddRemotesSomeUnreachable, - test s "return error when commit is locked" testCommitLock, test s "post commit that references an unknown proposal" testUnknownProposalRefCommit ], testGroup @@ -484,47 +482,6 @@ testAddRemotesSomeUnreachable = do memId (cmSelf (cnvMembers convAfter)) @?= alice cmOthers (cnvMembers convAfter) @?= [] -testCommitLock :: (HasCallStack) => TestM () -testCommitLock = do - users <- createAndConnectUsers (replicate 4 Nothing) - - runMLSTest $ do - [alice1, bob1, charlie1, dee1] <- traverse createMLSClient users - (groupId, _) <- setupMLSGroup alice1 - traverse_ uploadNewKeyPackage [bob1, charlie1, dee1] - - -- alice adds add bob - void $ createAddCommit alice1 [cidQualifiedUser bob1] >>= sendAndConsumeCommitBundle - - -- alice adds charlie - void $ createAddCommit alice1 [cidQualifiedUser charlie1] >>= sendAndConsumeCommitBundle - - -- simulate concurrent commit by blocking epoch - casClient <- view tsCass - runClient casClient $ insertLock groupId (Epoch 2) - - -- commit should fail due to competing lock - do - commit <- createAddCommit alice1 [cidQualifiedUser dee1] - bundle <- createBundle commit - err <- - responseJsonError - =<< localPostCommitBundle alice1 bundle - TestM () testUnknownProposalRefCommit = do [alice, bob] <- createAndConnectUsers (replicate 2 Nothing)