diff --git a/changelog.d/5-internal/WPB-22954 b/changelog.d/5-internal/WPB-22954 new file mode 100644 index 00000000000..133c43a40bd --- /dev/null +++ b/changelog.d/5-internal/WPB-22954 @@ -0,0 +1 @@ +Migration of mls commit locks from cassandra to postgres diff --git a/charts/wire-server/templates/background-worker/configmap.yaml b/charts/wire-server/templates/background-worker/configmap.yaml index d4fe2a63202..b12498bccff 100644 --- a/charts/wire-server/templates/background-worker/configmap.yaml +++ b/charts/wire-server/templates/background-worker/configmap.yaml @@ -84,6 +84,7 @@ data: migrateConversationCodes: {{ .migrateConversationCodes }} migrateTeamFeatures: {{ .migrateTeamFeatures }} migrateDomainRegistration: {{ .migrateDomainRegistration }} + migrateMLSCommitLocks: {{ .migrateMLSCommitLocks }} migrationOptions: {{ toYaml .migrationOptions | indent 6 }} diff --git a/charts/wire-server/values.yaml b/charts/wire-server/values.yaml index ae71d850795..536f6e46b19 100644 --- a/charts/wire-server/values.yaml +++ b/charts/wire-server/values.yaml @@ -90,6 +90,7 @@ galley: teamFeatures: cassandra domainRegistration: cassandra user: cassandra + mlsCommitLocks: cassandra settings: httpPoolSize: 128 maxTeamSize: 10000 @@ -1017,6 +1018,10 @@ background-worker: # It's important to set `settings.postgresMigration.domainRegistration` to `migration-to-postgresql` # before starting the migration. migrateDomainRegistration: false + # This will start the migration of mls commit locks. + # It's important to set `settings.postgresMigration.mlsCommitLocks` to `migration-to-postgresql` + # before starting the migration. + migrateMLSCommitLocks: false backendNotificationPusher: pushBackoffMinWait: 10000 # in microseconds, so 10ms diff --git a/docs/src/developer/reference/config-options.md b/docs/src/developer/reference/config-options.md index 066637dbb07..b91f2cde0bb 100644 --- a/docs/src/developer/reference/config-options.md +++ b/docs/src/developer/reference/config-options.md @@ -2106,12 +2106,14 @@ galley: teamFeatures: postgresql domainRegistration: postgresql user: postgresql + mlsCommitLocks: postgresql background-worker: config: migrateConversations: false migrateConversationCodes: false migrateTeamFeatures: false migrateDomainRegistration: false + migrateMLSCommitLocks: false ``` #### Migration for existing installations @@ -2142,6 +2144,7 @@ The current settings and their background-worker flags are: - `conversationCodes` -> `migrateConversationCodes` - `teamFeatures` -> `migrateTeamFeatures` - `domainRegistration` -> `migrateDomainRegistration` +- `mlsCommitLocks` -> `migrateMLSCommitLocks` **Migration pattern per migration setting** @@ -2160,13 +2163,15 @@ The current settings and their background-worker flags are: conversation: migration-to-postgresql conversationCodes: migration-to-postgresql teamFeatures: migration-to-postgresql - domainRegistration: cassandra + domainRegistration: cassandra + mlsCommitLocks: cassandra background-worker: config: migrateConversations: false migrateConversationCodes: false migrateTeamFeatures: false - migrateDomainRegistration: false + migrateDomainRegistration: false + migrateMLSCommitLocks: false ``` This change should restart the affected pods, and new writes will follow the @@ -2180,7 +2185,8 @@ The current settings and their background-worker flags are: migrateConversations: true migrateConversationCodes: true migrateTeamFeatures: true - migrateDomainRegistration: true + migrateDomainRegistration: true + migrateMLSCommitLocks: true ``` During migration, Cassandra rows are not deleted. Writes and migration share @@ -2196,6 +2202,7 @@ The current settings and their background-worker flags are: - `conversationCodes`: `wire_conv_codes_migration_finished` - `teamFeatures`: `wire_team_features_migration_finished` - `domainRegistration`: `wire_domain_registration_migration_finished` + - `mlsCommitLocks`: `wire_mls_commit_locks_migration_finished` 3. Cut over reads and writes to PostgreSQL for the selected migration setting(s). This configuration must be used from now on for every new @@ -2208,13 +2215,15 @@ The current settings and their background-worker flags are: conversation: postgresql conversationCodes: postgresql teamFeatures: postgresql - domainRegistration: cassandra + domainRegistration: cassandra + mlsCommitLocks: cassandra background-worker: config: migrateConversations: false migrateConversationCodes: false migrateTeamFeatures: false - migrateDomainRegistration: false + migrateDomainRegistration: false + migrateMLSCommitLocks: false ``` **How to run migrations independently or in batches** diff --git a/hack/helm_vars/common.yaml.gotmpl b/hack/helm_vars/common.yaml.gotmpl index 2276355e2a9..313b4f339bf 100644 --- a/hack/helm_vars/common.yaml.gotmpl +++ b/hack/helm_vars/common.yaml.gotmpl @@ -19,6 +19,7 @@ conversationCodesStore: {{ $preferredStore }} teamFeaturesStore: {{ $preferredStore }} domainRegistration: {{ $preferredStore }} userStore: {{ $preferredStore }} +mlsCommitLocksStore: {{ $preferredStore }} {{- if (eq (env "UPLOAD_XML_S3_BASE_URL") "") }} uploadXml: {} diff --git a/hack/helm_vars/wire-server/values.yaml.gotmpl b/hack/helm_vars/wire-server/values.yaml.gotmpl index bd6cdadf2a8..37baa2dd8b3 100644 --- a/hack/helm_vars/wire-server/values.yaml.gotmpl +++ b/hack/helm_vars/wire-server/values.yaml.gotmpl @@ -306,6 +306,7 @@ galley: teamFeatures: {{ .Values.teamFeaturesStore }} domainRegistration: {{ .Values.domainRegistration }} user: {{ .Values.userStore }} + mlsCommitLocks: {{ .Values.mlsCommitLocksStore }} settings: maxConvAndTeamSize: 16 maxTeamSize: 32 diff --git a/integration/integration.cabal b/integration/integration.cabal index c36b9c4e730..d19f39643e5 100644 --- a/integration/integration.cabal +++ b/integration/integration.cabal @@ -180,6 +180,7 @@ library Test.Migration.Conversation Test.Migration.ConversationCodes Test.Migration.DomainRegistration + Test.Migration.MLSCommitLock Test.Migration.TeamFeatures Test.Migration.Util Test.MLS diff --git a/integration/test/Test/Migration/MLSCommitLock.hs b/integration/test/Test/Migration/MLSCommitLock.hs new file mode 100644 index 00000000000..7c3ccaef23c --- /dev/null +++ b/integration/test/Test/Migration/MLSCommitLock.hs @@ -0,0 +1,74 @@ +-- 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.Migration.MLSCommitLock where + +import Control.Monad.Codensity +import Control.Monad.Reader +import MLS.Util +import SetupHelpers +import Test.Migration.Util (waitForMigration) +import Testlib.Prelude +import Testlib.ResourcePool + +-- | Verifies the MLS commit-lock store migration end to end. Every MLS commit +-- acquires and releases the commit lock, so driving commits through the three +-- storage locations exercises the lock in Cassandra, the dual-write mirror, and +-- Postgres-only. +testMLSCommitLockMigration :: (HasCallStack) => App () +testMLSCommitLockMigration = do + resourcePool <- asks (.resourcePool) + runCodensity (acquireResources 1 resourcePool) $ \[backend] -> do + let domain = backend.berDomain + + -- Cassandra: create an MLS group and commit once. This acquires and + -- releases the commit lock against Cassandra. + (alice1, convId) <- runCodensity (startDynamicBackend backend (conf "cassandra" False)) $ \_ -> do + alice <- randomUser domain def + alice1 <- createMLSClient def alice + bob <- randomUser domain def + bob1 <- createMLSClient def bob + void $ uploadNewKeyPackage def bob1 + convId <- createNewGroup def alice1 + void $ createAddCommit alice1 convId [bob] >>= sendAndConsumeCommitBundle + pure (alice1, convId) + + -- Dual-write + backfill: a commit is mirrored to Postgres, and the worker + -- copies any live locks until it reports completion. + runCodensity (startDynamicBackend backend (conf "migration-to-postgresql" True)) $ \_ -> do + charlie <- randomUser domain def + charlie1 <- createMLSClient def charlie + void $ uploadNewKeyPackage def charlie1 + void $ createAddCommit alice1 convId [charlie] >>= sendAndConsumeCommitBundle + waitForMigration domain counterName + + -- Postgres-only: a commit acquires and releases the lock against Postgres. + runCodensity (startDynamicBackend backend (conf "postgresql" False)) $ \_ -> do + dave <- randomUser domain def + dave1 <- createMLSClient def dave + void $ uploadNewKeyPackage def dave1 + void $ createAddCommit alice1 convId [dave] >>= sendAndConsumeCommitBundle + +conf :: String -> Bool -> ServiceOverrides +conf db runMigration = + def + { galleyCfg = setField "postgresMigration.mlsCommitLocks" db, + backgroundWorkerCfg = setField "migrateMLSCommitLocks" runMigration + } + +counterName :: String +counterName = "^wire_mls_commit_locks_migration_finished" diff --git a/libs/wire-subsystems/postgres-migrations/20260804143320-mls-commit-locks.sql b/libs/wire-subsystems/postgres-migrations/20260804143320-mls-commit-locks.sql new file mode 100644 index 00000000000..58d901b8047 --- /dev/null +++ b/libs/wire-subsystems/postgres-migrations/20260804143320-mls-commit-locks.sql @@ -0,0 +1,6 @@ +CREATE TABLE mls_commit_locks ( + group_id bytea NOT NULL, + epoch bigint NOT NULL, + expires_at timestamptz NOT NULL, + PRIMARY KEY (group_id, epoch) +); diff --git a/libs/wire-subsystems/src/Wire/ConversationStore/Cassandra/Queries.hs b/libs/wire-subsystems/src/Wire/ConversationStore/Cassandra/Queries.hs index fef06f288bb..11cf1afb8d6 100644 --- a/libs/wire-subsystems/src/Wire/ConversationStore/Cassandra/Queries.hs +++ b/libs/wire-subsystems/src/Wire/ConversationStore/Cassandra/Queries.hs @@ -360,6 +360,8 @@ acquireCommitLock = "insert into mls_commit_locks (group_id, epoch) values (?, ? releaseCommitLock :: PrepQuery W (GroupId, Epoch) () releaseCommitLock = "delete from mls_commit_locks where group_id = ? and epoch = ?" +selectAllCommitLocks :: PrepQuery R () (GroupId, Epoch) +selectAllCommitLocks = "select group_id, epoch from mls_commit_locks" -- Bots --------------------------------------------------------------------- diff --git a/libs/wire-subsystems/src/Wire/MLSCommitLockStore/DualWrite.hs b/libs/wire-subsystems/src/Wire/MLSCommitLockStore/DualWrite.hs new file mode 100644 index 00000000000..21d916af3fb --- /dev/null +++ b/libs/wire-subsystems/src/Wire/MLSCommitLockStore/DualWrite.hs @@ -0,0 +1,55 @@ +-- 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.DualWrite + ( interpretMLSCommitLockStoreToCassandraAndPostgres, + ) +where + +import Cassandra (ClientState) +import Imports +import Polysemy +import Polysemy.TinyLog (TinyLog) +import Wire.ConversationStore (LockAcquired (..), MLSCommitLockStore (..)) +import Wire.ConversationStore qualified as CommitLockStore +import Wire.ConversationStore.Cassandra qualified as Cassandra +import Wire.MLSCommitLockStore.Postgres qualified as Postgres +import Wire.Postgres (PGConstraints) + +-- | During migration Cassandra stays the source of truth: every write is +-- mirrored to Postgres, and 'AcquireCommitLock' returns the Cassandra result +-- (the arbiter) so mutual exclusion is governed by a single store until the +-- cutover to 'PostgresqlStorage'. +interpretMLSCommitLockStoreToCassandraAndPostgres :: + ( Member TinyLog r, + PGConstraints r + ) => + ClientState -> + InterpreterFor MLSCommitLockStore r +interpretMLSCommitLockStoreToCassandraAndPostgres client = interpret $ \case + AcquireCommitLock gId epoch ttl -> do + -- Cassandra is the arbiter: mirror the acquire to Postgres only when it + -- succeeds, so Postgres never holds a lock Cassandra did not grant. + acquired <- Cassandra.interpretMLSCommitLockStoreToCassandra client $ CommitLockStore.acquireCommitLock gId epoch ttl + when (acquired == Acquired) $ + void $ + Postgres.interpretMLSCommitLockStoreToPostgres $ + CommitLockStore.acquireCommitLock gId epoch ttl + pure acquired + ReleaseCommitLock gId epoch -> do + Cassandra.interpretMLSCommitLockStoreToCassandra client $ CommitLockStore.releaseCommitLock gId epoch + Postgres.interpretMLSCommitLockStoreToPostgres $ CommitLockStore.releaseCommitLock gId epoch diff --git a/libs/wire-subsystems/src/Wire/MLSCommitLockStore/Migration.hs b/libs/wire-subsystems/src/Wire/MLSCommitLockStore/Migration.hs new file mode 100644 index 00000000000..0e14b4d8dbb --- /dev/null +++ b/libs/wire-subsystems/src/Wire/MLSCommitLockStore/Migration.hs @@ -0,0 +1,161 @@ +-- 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.Migration (migrateMLSCommitLocksLoop) where + +import Cassandra hiding (Value) +import Data.Conduit +import Data.Conduit.List qualified as C +import Data.IORef qualified as IORef +import Data.Text qualified as T +import Data.Time +import Hasql.Pool.Extended qualified as Hasql +import Imports +import Polysemy +import Polysemy.Async +import Polysemy.Conc (interpretRace) +import Polysemy.Conc qualified as Conc +import Polysemy.Conc.Effect.Race hiding (Timeout) +import Polysemy.Input +import Polysemy.Resource (Resource, bracket, resourceToIOFinal) +import Polysemy.State +import Polysemy.TinyLog +import Prometheus qualified +import System.Logger qualified as Log +import UnliftIO qualified +import Wire.API.MLS.Epoch (Epoch) +import Wire.API.MLS.Group (GroupId, unGroupId) +import Wire.ConversationStore qualified as CommitLockStore +import Wire.ConversationStore.Cassandra.Queries qualified as Cql +import Wire.Migration +import Wire.MLSCommitLockStore.Postgres qualified as Postgres +import Wire.Postgres (PGConstraints) +import Wire.Sem.Logger (mapLogger) +import Wire.Sem.Logger.TinyLog (loggerToTinyLog) + +type EffectStack = + [ State Int, + Input ClientState, + Input Hasql.Pool, + Resource, + Async, + Race, + TinyLog, + Embed IO, + Final IO + ] + +migrateMLSCommitLocksLoop :: + MigrationOptions -> + ClientState -> + Hasql.Pool -> + Log.Logger -> + Prometheus.Counter -> + Prometheus.Counter -> + Prometheus.Counter -> + Prometheus.Vector Text Prometheus.Histogram -> + IO () +migrateMLSCommitLocksLoop migOpts cassClient pgPool logger migCounter migFinished migFailed migDuration = + migrationLoop + logger + "mls commit locks" + migFinished + migFailed + (interpreter cassClient pgPool logger "mls commit locks") + (migrateAllCommitLocks migOpts migCounter migDuration) + +interpreter :: ClientState -> Hasql.Pool -> Log.Logger -> ByteString -> Sem EffectStack a -> IO (Int, a) +interpreter cassClient pgPool logger name = + runFinal + . embedToFinal + . loggerToTinyLog logger + . mapLogger (Log.field "migration" (Log.val name) .) + . raiseUnder + . interpretRace + . asyncToIOFinal + . resourceToIOFinal + . runInputConst pgPool + . runInputConst cassClient + . runState 0 + +migrateAllCommitLocks :: + ( Member (Input Hasql.Pool) r, + Member (Embed IO) r, + Member (Input ClientState) r, + Member TinyLog r, + Member (State Int) r, + Member Resource r, + Member Race r + ) => + MigrationOptions -> + Prometheus.Counter -> + Prometheus.Vector Text Prometheus.Histogram -> + ConduitM () Void (Sem r) () +migrateAllCommitLocks migOpts migCounter migDuration = do + lift $ info $ Log.msg (Log.val "migrateAllCommitLocks") + withCount (paginateSem Cql.selectAllCommitLocks (paramsP LocalQuorum () migOpts.pageSize) x5) + .| logRetrievedPage migOpts.pageSize id + .| C.mapM_ (traverse_ (\row@(gId, _) -> handleErrors (unGroupId gId) (migrateCommitLockRow migOpts migCounter migDuration row))) + +-- | The lifetime an acquired commit lock is given. Cassandra auto-purges expired +-- rows, so every row read by the migration is live; we copy it with the same +-- lifetime the runtime uses (see 'withCommitLock' in +-- Wire.ConversationSubsystem.MLS.Util). +commitLockMigrationTtl :: NominalDiffTime +commitLockMigrationTtl = fromIntegral (600 :: Int) + +migrateCommitLockRow :: + ( PGConstraints r, + Member TinyLog r, + Member Resource r, + Member Race r + ) => + MigrationOptions -> + Prometheus.Counter -> + Prometheus.Vector Text Prometheus.Histogram -> + (GroupId, Epoch) -> + Sem r () +migrateCommitLockRow migOpts migCounter migDuration (gId, epoch) = + do + let keyText = T.pack (show gId) + outcomeRef <- liftIO $ IORef.newIORef @Text "error" + bracket + (liftIO getCurrentTime) + (observeDuration migDuration outcomeRef) + ( const $ do + timeoutResult <- Conc.timeout (migOpts.timeout <$ handleTimeout) migOpts.timeout $ Postgres.interpretMLSCommitLockStoreToPostgres $ CommitLockStore.acquireCommitLock gId epoch commitLockMigrationTtl + case timeoutResult of + Left timedOutAfter -> do + markOutcome outcomeRef "timeout" + liftIO . UnliftIO.throwIO $ MigrationTimedOut keyText timedOutAfter + Right _ -> do + markOutcome outcomeRef "success" + liftIO $ Prometheus.incCounter migCounter + ) + where + handleTimeout = + err $ + Log.msg (Log.val "mls commit lock migration timed out") + . Log.field "group_id" (show gId) + . Log.field "timeout" (show migOpts.timeout) + + markOutcome ref outcome = liftIO $ IORef.writeIORef ref outcome + + observeDuration metric outcomeRef start = do + outcome <- liftIO $ IORef.readIORef outcomeRef + end <- liftIO getCurrentTime + liftIO $ Prometheus.withLabel metric outcome (`Prometheus.observe` realToFrac (diffUTCTime end start)) 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..dbc627812f9 --- /dev/null +++ b/libs/wire-subsystems/src/Wire/MLSCommitLockStore/Postgres.hs @@ -0,0 +1,84 @@ +-- 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 Hasql.Statement qualified as Hasql +import Hasql.TH +import Imports +import Polysemy +import Wire.API.MLS.Epoch (Epoch) +import Wire.API.MLS.Group (GroupId) +import Wire.API.PostgresMarshall (dimapPG, lmapPG) +import Wire.ConversationStore (LockAcquired (..), MLSCommitLockStore (..)) +import Wire.Postgres (PGConstraints, runStatement) + +-- | Postgres interpreter for 'MLSCommitLockStore'. +-- +-- Acquire replicates Cassandra's @INSERT ... IF NOT EXISTS USING TTL@ as an +-- @INSERT ... ON CONFLICT DO UPDATE ... WHERE expires_at < now() RETURNING@: +-- +-- * no existing row -> INSERT succeeds -> 'Acquired' +-- * existing row, still live -> WHERE is false, no return -> 'NotAcquired' +-- * existing row, expired -> UPDATE succeeds -> 'Acquired' +-- +-- The last case is essential: unlike Cassandra (which purges expired TTL rows), +-- Postgres keeps the dead row, so we must treat an expired lock as re-acquirable +-- or a crashed holder would block its @(group_id, epoch)@ forever. +-- +-- Unlike Cassandra, Postgres never auto-purges expired rows, but the expired +-- branch above /reuses/ the existing row in place (UPDATE rather than INSERT), +-- so a re-acquired @(group_id, epoch)@ does not accumulate a second row. +-- Successful commits delete their row on release; only commits whose holder +-- crashed before release leave a dead row, which is unaddressable by future +-- commits (epochs are monotonic) and self-expires via @expires_at@. If dead-row +-- growth ever becomes operationally significant, a periodic +-- @DELETE FROM mls_commit_locks WHERE expires_at < now()@ (plus an index on +-- @expires_at@) can be added. +interpretMLSCommitLockStoreToPostgres :: + (PGConstraints r) => + InterpreterFor MLSCommitLockStore r +interpretMLSCommitLockStoreToPostgres = interpret $ \case + AcquireCommitLock gId epoch ttl -> do + let ttlSecs = round ttl :: Int32 + acquired <- runStatement (gId, epoch, ttlSecs) acquireStmt + pure $ maybe NotAcquired (const Acquired) acquired + ReleaseCommitLock gId epoch -> + runStatement (gId, epoch) releaseStmt + +acquireStmt :: Hasql.Statement (GroupId, Epoch, Int32) (Maybe Bool) +acquireStmt = + dimapPG + [maybeStatement| + INSERT INTO mls_commit_locks (group_id, epoch, expires_at) + VALUES ($1 :: bytea, $2 :: int8, now() + make_interval(secs => $3 :: int4)) + ON CONFLICT (group_id, epoch) DO UPDATE + SET expires_at = excluded.expires_at + WHERE mls_commit_locks.expires_at < now() + RETURNING true :: bool + |] + +releaseStmt :: Hasql.Statement (GroupId, Epoch) () +releaseStmt = + lmapPG + [resultlessStatement| + DELETE FROM mls_commit_locks + WHERE group_id = ($1 :: bytea) AND epoch = ($2 :: int8) + |] diff --git a/libs/wire-subsystems/src/Wire/PostgresMigrationOpts.hs b/libs/wire-subsystems/src/Wire/PostgresMigrationOpts.hs index 327862f7cd5..fa12ac9e8bc 100644 --- a/libs/wire-subsystems/src/Wire/PostgresMigrationOpts.hs +++ b/libs/wire-subsystems/src/Wire/PostgresMigrationOpts.hs @@ -56,7 +56,8 @@ data PostgresMigrationOpts = PostgresMigrationOpts conversationCodes :: StorageLocation, teamFeatures :: StorageLocation, domainRegistration :: StorageLocation, - user :: StorageLocation + user :: StorageLocation, + mlsCommitLocks :: StorageLocation } deriving (Show) @@ -68,3 +69,4 @@ instance FromJSON PostgresMigrationOpts where <*> o .: "teamFeatures" <*> o .: "domainRegistration" <*> o .: "user" + <*> o .: "mlsCommitLocks" diff --git a/libs/wire-subsystems/wire-subsystems.cabal b/libs/wire-subsystems/wire-subsystems.cabal index 943f8fa0294..ec44d4e758c 100644 --- a/libs/wire-subsystems/wire-subsystems.cabal +++ b/libs/wire-subsystems/wire-subsystems.cabal @@ -402,6 +402,9 @@ library Wire.MeetingsSubsystem.Notification Wire.Migration Wire.MigrationLock + Wire.MLSCommitLockStore.DualWrite + Wire.MLSCommitLockStore.Migration + Wire.MLSCommitLockStore.Postgres Wire.MlsKeyPackageStore Wire.MlsKeyPackageStore.Cassandra Wire.MlsKeyPackageSubsystem diff --git a/postgres-schema.sql b/postgres-schema.sql index 948f1d62594..75fd93e5396 100644 --- a/postgres-schema.sql +++ b/postgres-schema.sql @@ -9,8 +9,8 @@ \restrict 79bbfb4630959c48307653a5cd3d83f2582b3c2210f75f10d79e3ebf0015620 --- Dumped from database version 17.9 --- Dumped by pg_dump version 17.9 +-- Dumped from database version 17.10 +-- Dumped by pg_dump version 17.10 SET statement_timeout = 0; SET lock_timeout = 0; @@ -312,6 +312,19 @@ CREATE TABLE public.meetings ( ALTER TABLE public.meetings OWNER TO "wire-server"; +-- +-- Name: mls_commit_locks; Type: TABLE; Schema: public; Owner: wire-server +-- + +CREATE TABLE public.mls_commit_locks ( + group_id bytea NOT NULL, + epoch bigint NOT NULL, + expires_at timestamp with time zone NOT NULL +); + + +ALTER TABLE public.mls_commit_locks OWNER TO "wire-server"; + -- -- Name: mls_group_member_client; Type: TABLE; Schema: public; Owner: wire-server -- @@ -585,6 +598,14 @@ ALTER TABLE ONLY public.meetings ADD CONSTRAINT meetings_pkey PRIMARY KEY (id); +-- +-- Name: mls_commit_locks mls_commit_locks_pkey; Type: CONSTRAINT; Schema: public; Owner: wire-server +-- + +ALTER TABLE ONLY public.mls_commit_locks + ADD CONSTRAINT mls_commit_locks_pkey PRIMARY KEY (group_id, epoch); + + -- -- Name: mls_group_member_client mls_group_member_client_pkey; Type: CONSTRAINT; Schema: public; Owner: wire-server -- diff --git a/services/background-worker/background-worker.integration.yaml b/services/background-worker/background-worker.integration.yaml index e264ce14016..7571d1bc3b5 100644 --- a/services/background-worker/background-worker.integration.yaml +++ b/services/background-worker/background-worker.integration.yaml @@ -92,3 +92,4 @@ postgresMigration: teamFeatures: postgresql domainRegistration: postgresql user: postgresql + mlsCommitLocks: postgresql diff --git a/services/background-worker/src/Wire/BackgroundWorker.hs b/services/background-worker/src/Wire/BackgroundWorker.hs index b57ba12df40..937ee8b4575 100644 --- a/services/background-worker/src/Wire/BackgroundWorker.hs +++ b/services/background-worker/src/Wire/BackgroundWorker.hs @@ -78,6 +78,13 @@ run opts galleyOpts = do withNamedLogger "migrate-domain-registration" $ Migrations.domainRegistration opts.migrationOptions else pure $ pure () + cleanupMLSLocksMigration <- + if opts.migrateMLSCommitLocks + then + runAppT env $ + withNamedLogger "migrate-mls-commit-locks" $ + Migrations.mlsCommitLocks opts.migrationOptions + else pure $ pure () cleanupJobs <- runAppT env $ withNamedLogger "background-job-consumer" $ @@ -89,13 +96,14 @@ run opts galleyOpts = do let cleanup = void $ runConcurrently $ - (,,,,,,,) + (,,,,,,,,) <$> Concurrently cleanupDeadUserNotifWatcher <*> Concurrently cleanupBackendNotifPusher <*> Concurrently cleanupConvMigration <*> Concurrently cleanUpConvCodesMigration <*> Concurrently cleanupTeamFeaturesMigration <*> Concurrently cleanupDomainRegistrationMigration + <*> Concurrently cleanupMLSLocksMigration <*> Concurrently cleanupJobRunner <*> Concurrently cleanupJobs diff --git a/services/background-worker/src/Wire/BackgroundWorker/Options.hs b/services/background-worker/src/Wire/BackgroundWorker/Options.hs index 61df5d5d14f..f75fcbb8b35 100644 --- a/services/background-worker/src/Wire/BackgroundWorker/Options.hs +++ b/services/background-worker/src/Wire/BackgroundWorker/Options.hs @@ -55,6 +55,7 @@ data Opts = Opts migrateConversationCodes :: !Bool, migrateTeamFeatures :: !Bool, migrateDomainRegistration :: !Bool, + migrateMLSCommitLocks :: !Bool, jobs :: JobConfig, meetingsCleanup :: MeetingsCleanupConfig, backgroundJobs :: BackgroundJobsConfig diff --git a/services/background-worker/src/Wire/PostgresMigrations.hs b/services/background-worker/src/Wire/PostgresMigrations.hs index 604cab0140c..3c3f64fdc1e 100644 --- a/services/background-worker/src/Wire/PostgresMigrations.hs +++ b/services/background-worker/src/Wire/PostgresMigrations.hs @@ -26,6 +26,7 @@ import Wire.BackgroundWorker.Util import Wire.CodeStore.Migration import Wire.ConversationStore.Migration import Wire.DomainRegistrationStore.Migration +import Wire.MLSCommitLockStore.Migration import Wire.Migration (MigrationOptions) import Wire.TeamFeatureStore.Migration @@ -107,3 +108,21 @@ domainRegistration migOpts = do pure $ do Log.info logger $ Log.msg (Log.val "cancelling domain registration migration") cancel migrationLoop + +mlsCommitLocks :: MigrationOptions -> AppT IO CleanupAction +mlsCommitLocks migOpts = do + cassClient <- asks (.cassandraGalley) + pgPool <- asks (.hasqlPool) + logger <- asks (.logger) + Log.info logger $ Log.msg (Log.val "starting mls commit locks migration") + count <- register $ counter $ Prometheus.Info "wire_mls_commit_locks_migrated_to_pg" "Number of mls commit locks migrated to Postgresql" + finished <- register $ counter $ Prometheus.Info "wire_mls_commit_locks_migration_finished" "Whether the mls commit locks migration to Postgresql is finished successfully" + failed <- register $ counter $ Prometheus.Info "wire_mls_commit_locks_migration_failed" "Whether the mls commit locks migration to Postgresql has failed" + duration <- register $ vector "outcome" $ histogram (Prometheus.Info "wire_mls_commit_locks_migration_duration_seconds" "Duration of mls commit lock migration attempts") defaultBuckets + + migrationLoop <- async . lift $ migrateMLSCommitLocksLoop migOpts cassClient pgPool logger count finished failed duration + + Log.info logger $ Log.msg (Log.val "started mls commit locks migration") + pure $ do + Log.info logger $ Log.msg (Log.val "cancelling mls commit locks migration") + cancel migrationLoop diff --git a/services/background-worker/test/Test/Wire/BackendNotificationPusherSpec.hs b/services/background-worker/test/Test/Wire/BackendNotificationPusherSpec.hs index 7222120d93a..421ce3f91c3 100644 --- a/services/background-worker/test/Test/Wire/BackendNotificationPusherSpec.hs +++ b/services/background-worker/test/Test/Wire/BackendNotificationPusherSpec.hs @@ -371,7 +371,8 @@ spec = do conversationCodes = CassandraStorage, teamFeatures = CassandraStorage, domainRegistration = CassandraStorage, - user = CassandraStorage + user = CassandraStorage, + mlsCommitLocks = CassandraStorage } gundeckEndpoint = undefined brigEndpoint = undefined @@ -435,7 +436,8 @@ spec = do conversationCodes = CassandraStorage, teamFeatures = CassandraStorage, domainRegistration = CassandraStorage, - user = CassandraStorage + user = CassandraStorage, + mlsCommitLocks = CassandraStorage } gundeckEndpoint = undefined brigEndpoint = undefined diff --git a/services/background-worker/test/Test/Wire/Util.hs b/services/background-worker/test/Test/Wire/Util.hs index 5d89532bfec..f4511334905 100644 --- a/services/background-worker/test/Test/Wire/Util.hs +++ b/services/background-worker/test/Test/Wire/Util.hs @@ -50,7 +50,8 @@ testEnv = do conversationCodes = CassandraStorage, teamFeatures = CassandraStorage, domainRegistration = CassandraStorage, - user = CassandraStorage + user = CassandraStorage, + mlsCommitLocks = CassandraStorage } statuses <- newIORef mempty backendNotificationMetrics <- mkBackendNotificationMetrics diff --git a/services/brig/brig.integration.yaml b/services/brig/brig.integration.yaml index 8be11f028bd..c428d24068a 100644 --- a/services/brig/brig.integration.yaml +++ b/services/brig/brig.integration.yaml @@ -176,6 +176,7 @@ postgresMigration: teamFeatures: postgresql domainRegistration: postgresql user: postgresql + mlsCommitLocks: postgresql optSettings: setActivationTimeout: 4 diff --git a/services/galley/galley.integration.yaml b/services/galley/galley.integration.yaml index 47980040c1e..ae1cafe52dd 100644 --- a/services/galley/galley.integration.yaml +++ b/services/galley/galley.integration.yaml @@ -266,3 +266,4 @@ postgresMigration: teamFeatures: postgresql domainRegistration: postgresql user: postgresql + mlsCommitLocks: postgresql diff --git a/services/galley/src/Galley/App.hs b/services/galley/src/Galley/App.hs index bd0131b4d2d..6241b2f1577 100644 --- a/services/galley/src/Galley/App.hs +++ b/services/galley/src/Galley/App.hs @@ -107,6 +107,8 @@ import Wire.CodeStore.DualWrite import Wire.CodeStore.Postgres import Wire.ConversationStore (ConversationStore, MLSCommitLockStore) import Wire.ConversationStore.Cassandra (MigrationError (..), interpretConversationStoreByMigration, interpretMLSCommitLockStoreToCassandra) +import Wire.MLSCommitLockStore.DualWrite (interpretMLSCommitLockStoreToCassandraAndPostgres) +import Wire.MLSCommitLockStore.Postgres (interpretMLSCommitLockStoreToPostgres) import Wire.ConversationSubsystem import Wire.ConversationSubsystem.Interpreter (ConversationSubsystemError, GroupInfoCheckEnabled (..), IntraListing (IntraListing), interpretConversationSubsystem) import Wire.CustomBackendStore @@ -423,6 +425,11 @@ evalGalley e = CassandraStorage -> interpretTeamFeatureStoreToCassandra MigrationToPostgresql -> interpretTeamFeatureStoreToCassandraAndPostgres PostgresqlStorage -> interpretTeamFeatureStoreToPostgres + mlsCommitLockStoreInterpreter = + case (e ^. options . postgresMigration).mlsCommitLocks of + CassandraStorage -> interpretMLSCommitLockStoreToCassandra (e ^. cstate) + MigrationToPostgresql -> interpretMLSCommitLockStoreToCassandraAndPostgres (e ^. cstate) + PostgresqlStorage -> interpretMLSCommitLockStoreToPostgres localUnit = toLocalUnsafe (e ^. options . settings . federationDomain) () teamSubsystemConfig = TeamSubsystemConfig @@ -525,7 +532,7 @@ evalGalley e = . interpretTeamMemberStoreToCassandraWithPaging lh . interpretTeamMemberStoreToCassandra lh . teamFeatureStoreInterpreter - . interpretMLSCommitLockStoreToCassandra (e ^. cstate) + . mlsCommitLockStoreInterpreter . convStoreInterpreter . interpretTeamNotificationStoreToCassandra . interpretServiceStoreToCassandra (e ^. cstate)