From d79a73f0162d1c620d75f79cca81362ede91aaed Mon Sep 17 00:00:00 2001 From: Gautier DI FOLCO Date: Tue, 4 Aug 2026 10:21:21 +0200 Subject: [PATCH] WPB-22955: migrate mls-proposal-refs to PostGreSQL Migrate the Wire.ProposalStore (MLS pending proposal refs, table mls_proposal_refs) from Cassandra to PostgreSQL using the dual-write + background-worker cutover pattern, mirroring Wire.CodeStore. Adds Postgres + DualWrite interpreters selected at startup by postgresMigration.proposals, a TTL-preserving backfill loop in the background worker, the SQL migration + regenerated postgres-schema.sql, helm/gotmpl/docs config, and an integration test driving the cassandra -> migration-to-postgresql -> postgresql lifecycle. Roundtrip properties cover the ProposalRef/ProposalOrigin/RawMLS Proposal marshall instances. --- changelog.d/5-internal/WPB-22955 | 1 + .../background-worker/configmap.yaml | 1 + charts/wire-server/values.yaml | 5 + .../src/developer/reference/config-options.md | 8 + hack/helm_vars/common.yaml.gotmpl | 1 + hack/helm_vars/wire-server/values.yaml.gotmpl | 1 + integration/integration.cabal | 1 + integration/test/Test/Migration/Proposals.hs | 71 +++++++ libs/wire-api/src/Wire/API/MLS/Proposal.hs | 24 ++- .../Wire/API/Roundtrip/PostgresMarshall.hs | 7 +- .../20260804073531-mls-proposal-refs.sql | 13 ++ .../src/Wire/PostgresMigrationOpts.hs | 4 +- .../src/Wire/ProposalStore/Cassandra.hs | 4 + .../src/Wire/ProposalStore/DualWrite.hs | 52 ++++++ .../src/Wire/ProposalStore/Migration.hs | 176 ++++++++++++++++++ .../src/Wire/ProposalStore/Postgres.hs | 148 +++++++++++++++ libs/wire-subsystems/wire-subsystems.cabal | 3 + postgres-schema.sql | 31 +++ .../background-worker.integration.yaml | 2 + .../src/Wire/BackgroundWorker.hs | 10 +- .../src/Wire/BackgroundWorker/Options.hs | 1 + .../src/Wire/PostgresMigrations.hs | 19 ++ .../Wire/BackendNotificationPusherSpec.hs | 6 +- .../background-worker/test/Test/Wire/Util.hs | 3 +- services/brig/brig.integration.yaml | 1 + services/galley/galley.integration.yaml | 1 + services/galley/src/Galley/App.hs | 9 +- 27 files changed, 595 insertions(+), 8 deletions(-) create mode 100644 changelog.d/5-internal/WPB-22955 create mode 100644 integration/test/Test/Migration/Proposals.hs create mode 100644 libs/wire-subsystems/postgres-migrations/20260804073531-mls-proposal-refs.sql create mode 100644 libs/wire-subsystems/src/Wire/ProposalStore/DualWrite.hs create mode 100644 libs/wire-subsystems/src/Wire/ProposalStore/Migration.hs create mode 100644 libs/wire-subsystems/src/Wire/ProposalStore/Postgres.hs diff --git a/changelog.d/5-internal/WPB-22955 b/changelog.d/5-internal/WPB-22955 new file mode 100644 index 00000000000..c1a8f044464 --- /dev/null +++ b/changelog.d/5-internal/WPB-22955 @@ -0,0 +1 @@ +Migration of mls_proposal_refs 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..fc21b0760fd 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 }} + migrateProposals: {{ .migrateProposals }} migrationOptions: {{ toYaml .migrationOptions | indent 6 }} diff --git a/charts/wire-server/values.yaml b/charts/wire-server/values.yaml index ae71d850795..746969356f8 100644 --- a/charts/wire-server/values.yaml +++ b/charts/wire-server/values.yaml @@ -90,6 +90,7 @@ galley: teamFeatures: cassandra domainRegistration: cassandra user: cassandra + proposals: 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 proposal refs. + # It's important to set `settings.postgresMigration.proposals` to `migration-to-postgresql` + # before starting the migration. + migrateProposals: 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..09a7f918d8c 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 + proposals: postgresql background-worker: config: migrateConversations: false migrateConversationCodes: false migrateTeamFeatures: false migrateDomainRegistration: false + migrateProposals: false ``` #### Migration for existing installations @@ -2142,6 +2144,7 @@ The current settings and their background-worker flags are: - `conversationCodes` -> `migrateConversationCodes` - `teamFeatures` -> `migrateTeamFeatures` - `domainRegistration` -> `migrateDomainRegistration` +- `proposals` -> `migrateProposals` **Migration pattern per migration setting** @@ -2161,12 +2164,14 @@ The current settings and their background-worker flags are: conversationCodes: migration-to-postgresql teamFeatures: migration-to-postgresql domainRegistration: cassandra + proposals: cassandra background-worker: config: migrateConversations: false migrateConversationCodes: false migrateTeamFeatures: false migrateDomainRegistration: false + migrateProposals: false ``` This change should restart the affected pods, and new writes will follow the @@ -2196,6 +2201,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` + - `proposals`: `wire_mls_proposal_refs_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 @@ -2209,12 +2215,14 @@ The current settings and their background-worker flags are: conversationCodes: postgresql teamFeatures: postgresql domainRegistration: cassandra + proposals: cassandra background-worker: config: migrateConversations: false migrateConversationCodes: false migrateTeamFeatures: false migrateDomainRegistration: false + migrateProposals: 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..8c7cbf87afd 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 }} +proposalsStore: {{ $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..f89be73c33e 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 }} + proposals: {{ .Values.proposalsStore }} settings: maxConvAndTeamSize: 16 maxTeamSize: 32 diff --git a/integration/integration.cabal b/integration/integration.cabal index c36b9c4e730..2a4662c62b8 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.Proposals Test.Migration.TeamFeatures Test.Migration.Util Test.MLS diff --git a/integration/test/Test/Migration/Proposals.hs b/integration/test/Test/Migration/Proposals.hs new file mode 100644 index 00000000000..db6465cab5b --- /dev/null +++ b/integration/test/Test/Migration/Proposals.hs @@ -0,0 +1,71 @@ +module Test.Migration.Proposals where + +import API.Galley +import Control.Monad.Codensity +import Control.Monad.Reader +import MLS.Util +import SetupHelpers +import Test.Migration.Util (waitForMigration) +import Testlib.Prelude +import Testlib.ResourcePool + +-- | Migrates the MLS proposal store ('mls_proposal_refs') from Cassandra to +-- PostgreSQL. +-- +-- Drives the operator lifecycle across dynamic backends: @cassandra@ -> +-- @migration-to-postgresql@ (worker off) -> @migration-to-postgresql@ (worker on) +-- -> @postgresql@. It verifies that galley and the background worker boot under +-- each 'StorageLocation', that the worker runs to completion +-- (@'waitForMigration'@ reaches @1.0@), and that MLS group operation keeps +-- working after the cutover to PostgreSQL (internal-commit processing resolves +-- pending proposal references through @'getAllPendingProposals'@, now served by +-- the PostgreSQL interpreter). +-- +-- Note: this test does not assert backfill read-back fidelity of pre-existing +-- rows (pending proposals are short-lived and consumed by later commits, so none +-- reliably survive the cutover to read back). The bytea marshalling and the +-- TTL-preserving backfill upsert are covered by the 'PostgresMarshall' roundtrip +-- properties in @wire-api@ and by the proven @CodeStore@ upsert this migration +-- mirrors. +testProposalsMigration :: (HasCallStack) => App () +testProposalsMigration = do + resourcePool <- asks (.resourcePool) + runCodensity (acquireResources 1 resourcePool) $ \[backend] -> do + let domain = backend.berDomain + + (admin, convId, adminClient) <- + runCodensity (startDynamicBackend backend (conf "cassandra" False)) $ \_ -> do + (admin, _tid, _members) <- createTeam domain 2 + bob <- randomUser domain def + [adminClient, bobClient] <- traverse (createMLSClient def) [admin, bob] + void $ uploadNewKeyPackage def bobClient + convId <- createNewGroup def adminClient + void $ createAddCommit adminClient convId [bob] >>= sendAndConsumeCommitBundle + -- Store a pending (backend) proposal under Cassandra. + void $ createPendingProposalCommit convId adminClient >>= sendAndConsumeCommitBundle + pure (admin, convId, adminClient) + + runCodensity (startDynamicBackend backend (conf "migration-to-postgresql" False)) $ \_ -> do + -- New writes are dual-written; the previously stored proposal is still + -- Cassandra-only until the worker runs. + void $ createPendingProposalCommit convId adminClient >>= sendAndConsumeCommitBundle + + runCodensity (startDynamicBackend backend (conf "migration-to-postgresql" True)) $ \_ -> do + waitForMigration domain counterName + + runCodensity (startDynamicBackend backend (conf "postgresql" False)) $ \_ -> do + -- Reads now come from PostgreSQL; MLS commit processing resolves pending + -- proposal references through the migrated store. + void $ createPendingProposalCommit convId adminClient >>= sendAndConsumeCommitBundle + bindResponse (getConversation admin (convIdToQidObject convId)) $ \resp -> do + resp.status `shouldMatchInt` 200 + +conf :: String -> Bool -> ServiceOverrides +conf db runMigration = + def + { galleyCfg = setField "postgresMigration.proposals" db, + backgroundWorkerCfg = setField "migrateProposals" runMigration + } + +counterName :: String +counterName = "^wire_mls_proposal_refs_migration_finished" diff --git a/libs/wire-api/src/Wire/API/MLS/Proposal.hs b/libs/wire-api/src/Wire/API/MLS/Proposal.hs index 1ae2ef989ac..3031c9e1c50 100644 --- a/libs/wire-api/src/Wire/API/MLS/Proposal.hs +++ b/libs/wire-api/src/Wire/API/MLS/Proposal.hs @@ -22,8 +22,10 @@ module Wire.API.MLS.Proposal where import Cassandra import Control.Lens (makePrisms) +import Data.Bifunctor (first) import Data.Binary import Data.ByteString as B +import Data.Text qualified as Text import GHC.Records import Imports import Test.QuickCheck @@ -35,6 +37,7 @@ import Wire.API.MLS.LeafNode import Wire.API.MLS.ProposalTag import Wire.API.MLS.ProtocolVersion import Wire.API.MLS.Serialisation +import Wire.API.PostgresMarshall import Wire.Arbitrary -- | https://messaginglayersecurity.rocks/mls-protocol/draft-ietf-mls-protocol-20/draft-ietf-mls-protocol.html#section-12.1-2 @@ -252,7 +255,8 @@ makePrisms ''ProposalOrRef data ProposalOrigin = ProposalOriginClient | ProposalOriginBackend - deriving (Eq) + deriving (Eq, Show, Generic) + deriving (Arbitrary) via (GenericUniform ProposalOrigin) instance Cql ProposalOrigin where ctype = Tagged IntColumn @@ -268,3 +272,21 @@ intToOrigin :: Int32 -> Either String ProposalOrigin intToOrigin 0 = pure ProposalOriginClient intToOrigin 1 = pure ProposalOriginBackend intToOrigin n = Left $ "intToOrigin: unexptected int constant: " <> show n + +instance PostgresMarshall ByteString ProposalRef where + postgresMarshall = (.unProposalRef) + +instance PostgresUnmarshall ByteString ProposalRef where + postgresUnmarshall = Right . ProposalRef + +instance PostgresMarshall Int32 ProposalOrigin where + postgresMarshall = originToInt + +instance PostgresUnmarshall Int32 ProposalOrigin where + postgresUnmarshall = first Text.pack . intToOrigin + +instance PostgresMarshall ByteString (RawMLS Proposal) where + postgresMarshall = (.raw) + +instance PostgresUnmarshall ByteString (RawMLS Proposal) where + postgresUnmarshall = decodeMLS' diff --git a/libs/wire-api/test/unit/Test/Wire/API/Roundtrip/PostgresMarshall.hs b/libs/wire-api/test/unit/Test/Wire/API/Roundtrip/PostgresMarshall.hs index 191f55bffd0..79704b6f00f 100644 --- a/libs/wire-api/test/unit/Test/Wire/API/Roundtrip/PostgresMarshall.hs +++ b/libs/wire-api/test/unit/Test/Wire/API/Roundtrip/PostgresMarshall.hs @@ -29,6 +29,8 @@ import Imports import Test.Tasty qualified as T import Test.Tasty.QuickCheck import Type.Reflection (typeRep) +import Wire.API.MLS.Proposal (Proposal, ProposalOrigin, ProposalRef) +import Wire.API.MLS.Serialisation (RawMLS) import Wire.API.Password as Password import Wire.API.Password.Argon2id (Argon2HashedPassword (..), encodeArgon2HashedPassword) import Wire.API.Password.Scrypt (encodeScryptPassword) @@ -44,7 +46,10 @@ tests = testRoundTrip @ByteString @Password.Password, testRoundTrip @Int32 @FeatureStatus, testRoundTrip @Int32 @LockStatus, - testRoundTrip @A.Value @DbConfig + testRoundTrip @A.Value @DbConfig, + testRoundTrip @ByteString @ProposalRef, + testRoundTrip @Int32 @ProposalOrigin, + testRoundTrip @ByteString @(RawMLS Proposal) ] testRoundTrip :: diff --git a/libs/wire-subsystems/postgres-migrations/20260804073531-mls-proposal-refs.sql b/libs/wire-subsystems/postgres-migrations/20260804073531-mls-proposal-refs.sql new file mode 100644 index 00000000000..0a4c092244d --- /dev/null +++ b/libs/wire-subsystems/postgres-migrations/20260804073531-mls-proposal-refs.sql @@ -0,0 +1,13 @@ +CREATE TABLE IF NOT EXISTS mls_proposal_refs ( + group_id bytea NOT NULL, + epoch int8 NOT NULL, + ref bytea NOT NULL, + origin int4, + proposal bytea NOT NULL, + expires_at timestamptz NOT NULL, + PRIMARY KEY (group_id, epoch, ref) +); + +-- index for lookups like `WHERE group_id = ? AND epoch = ? AND expires_at > now()` +CREATE INDEX mls_proposal_refs_group_epoch_expires_at_idx + ON mls_proposal_refs (group_id, epoch, expires_at); diff --git a/libs/wire-subsystems/src/Wire/PostgresMigrationOpts.hs b/libs/wire-subsystems/src/Wire/PostgresMigrationOpts.hs index 327862f7cd5..5f4a3c3d31f 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, + proposals :: StorageLocation } deriving (Show) @@ -68,3 +69,4 @@ instance FromJSON PostgresMigrationOpts where <*> o .: "teamFeatures" <*> o .: "domainRegistration" <*> o .: "user" + <*> o .: "proposals" diff --git a/libs/wire-subsystems/src/Wire/ProposalStore/Cassandra.hs b/libs/wire-subsystems/src/Wire/ProposalStore/Cassandra.hs index aab1f480fea..0323b5febb8 100644 --- a/libs/wire-subsystems/src/Wire/ProposalStore/Cassandra.hs +++ b/libs/wire-subsystems/src/Wire/ProposalStore/Cassandra.hs @@ -17,6 +17,7 @@ module Wire.ProposalStore.Cassandra ( interpretProposalStoreToCassandra, + selectAllProposals, ProposalOrigin (..), ) where @@ -85,3 +86,6 @@ getAllPending = "select ref, origin, proposal from mls_proposal_refs where group deleteAllProposalsForGroup :: PrepQuery W (Identity GroupId) () deleteAllProposalsForGroup = "delete from mls_proposal_refs where group_id = ?" + +selectAllProposals :: PrepQuery R () (GroupId, Epoch, ProposalRef, Maybe ProposalOrigin, Int32, RawMLS Proposal) +selectAllProposals = "select group_id, epoch, ref, origin, ttl(proposal), proposal from mls_proposal_refs" diff --git a/libs/wire-subsystems/src/Wire/ProposalStore/DualWrite.hs b/libs/wire-subsystems/src/Wire/ProposalStore/DualWrite.hs new file mode 100644 index 00000000000..7c14b484533 --- /dev/null +++ b/libs/wire-subsystems/src/Wire/ProposalStore/DualWrite.hs @@ -0,0 +1,52 @@ +-- 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.ProposalStore.DualWrite + ( interpretProposalStoreToCassandraAndPostgres, + ) +where + +import Cassandra (ClientState) +import Imports +import Polysemy +import Polysemy.Input +import Wire.Postgres (PGConstraints) +import Wire.ProposalStore (ProposalStore (..)) +import Wire.ProposalStore qualified as ProposalStore +import Wire.ProposalStore.Cassandra qualified as Cassandra +import Wire.ProposalStore.Postgres qualified as Postgres + +-- | Cassandra is the source of truth during migration; writes are mirrored to Postgres. +interpretProposalStoreToCassandraAndPostgres :: + ( Member (Input ClientState) r, + PGConstraints r + ) => + Sem (ProposalStore ': r) a -> + Sem r a +interpretProposalStoreToCassandraAndPostgres = interpret $ \case + GetProposal groupId epoch ref -> + Cassandra.interpretProposalStoreToCassandra $ ProposalStore.getProposal groupId epoch ref + GetAllPendingProposalRefs groupId epoch -> + Cassandra.interpretProposalStoreToCassandra $ ProposalStore.getAllPendingProposalRefs groupId epoch + GetAllPendingProposals groupId epoch -> + Cassandra.interpretProposalStoreToCassandra $ ProposalStore.getAllPendingProposals groupId epoch + StoreProposal groupId epoch sp -> do + Cassandra.interpretProposalStoreToCassandra $ ProposalStore.storeProposal groupId epoch sp + Postgres.interpretProposalStoreToPostgres $ ProposalStore.storeProposal groupId epoch sp + DeleteAllProposals groupId -> do + Cassandra.interpretProposalStoreToCassandra $ ProposalStore.deleteAllProposals groupId + Postgres.interpretProposalStoreToPostgres $ ProposalStore.deleteAllProposals groupId diff --git a/libs/wire-subsystems/src/Wire/ProposalStore/Migration.hs b/libs/wire-subsystems/src/Wire/ProposalStore/Migration.hs new file mode 100644 index 00000000000..c236975ee1c --- /dev/null +++ b/libs/wire-subsystems/src/Wire/ProposalStore/Migration.hs @@ -0,0 +1,176 @@ +-- 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.ProposalStore.Migration (migrateProposalsLoop) 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 Hasql.Statement qualified as Hasql +import Hasql.TH +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 +import Wire.API.MLS.Group +import Wire.API.MLS.Proposal +import Wire.API.MLS.Serialisation +import Wire.API.PostgresMarshall +import Wire.Migration +import Wire.Postgres +import Wire.ProposalStore.Cassandra qualified as Cql +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 + ] + +migrateProposalsLoop :: + MigrationOptions -> + ClientState -> + Hasql.Pool -> + Log.Logger -> + Prometheus.Counter -> + Prometheus.Counter -> + Prometheus.Counter -> + Prometheus.Vector Text Prometheus.Histogram -> + IO () +migrateProposalsLoop migOpts cassClient pgPool logger migCounter migFinished migFailed migDuration = + migrationLoop + logger + "mls proposal refs" + migFinished + migFailed + (interpreter cassClient pgPool logger "mls proposal refs") + (migrateAllProposals 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 + +migrateAllProposals :: + ( 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) () +migrateAllProposals migOpts migCounter migDuration = do + lift $ info $ Log.msg (Log.val "migrateAllProposals") + withCount (paginateSem Cql.selectAllProposals (paramsP LocalQuorum () migOpts.pageSize) x5) + .| logRetrievedPage migOpts.pageSize id + .| C.mapM_ (traverse_ (\row@(groupId, _, _, _, _, _) -> handleErrors (unGroupId groupId) (migrateProposalRow migOpts migCounter migDuration row))) + +migrateProposalRow :: + ( PGConstraints r, + Member TinyLog r, + Member Resource r, + Member Race r + ) => + MigrationOptions -> + Prometheus.Counter -> + Prometheus.Vector Text Prometheus.Histogram -> + (GroupId, Epoch, ProposalRef, Maybe ProposalOrigin, Int32, RawMLS Proposal) -> + Sem r () +migrateProposalRow migOpts migCounter migDuration (groupId, epoch, ref, origin, ttl, proposal) = + when (ttl > 0) $ do + let keyText = T.pack (show groupId) + outcomeRef <- liftIO $ IORef.newIORef @Text "error" + bracket + (liftIO getCurrentTime) + (observeDuration migDuration outcomeRef) + ( const $ do + timeoutResult <- + Conc.timeout + (migOpts.timeout <$ handleTimeout) + migOpts.timeout + $ runStatement (groupId, epoch, ref, origin, proposal, ttl) upsert + 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 + upsert :: + Hasql.Statement (GroupId, Epoch, ProposalRef, Maybe ProposalOrigin, RawMLS Proposal, Int32) () + upsert = + lmapPG + [resultlessStatement| + INSERT INTO mls_proposal_refs (group_id, epoch, ref, origin, proposal, expires_at) + VALUES ($1 :: bytea, $2 :: int8, $3 :: bytea, $4 :: int4?, $5 :: bytea, now () + make_interval(secs => $6 :: int)) + ON CONFLICT (group_id, epoch, ref) DO UPDATE + SET origin = ($4 :: int4?), + proposal = ($5 :: bytea), + expires_at = now () + make_interval(secs => $6 :: int) + |] + + handleTimeout = + err $ + Log.msg (Log.val "mls proposal ref migration timed out") + . Log.field "group_id" (show groupId) + . Log.field "epoch" (show epoch) + . Log.field "ref" (show ref) + . 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/ProposalStore/Postgres.hs b/libs/wire-subsystems/src/Wire/ProposalStore/Postgres.hs new file mode 100644 index 00000000000..a7035f06da0 --- /dev/null +++ b/libs/wire-subsystems/src/Wire/ProposalStore/Postgres.hs @@ -0,0 +1,148 @@ +-- 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.ProposalStore.Postgres + ( interpretProposalStoreToPostgres, + ) +where + +import Hasql.Statement qualified as Hasql +import Hasql.TH +import Imports +import Polysemy +import Wire.API.MLS.Epoch +import Wire.API.MLS.Group +import Wire.API.MLS.Proposal +import Wire.API.MLS.Serialisation +import Wire.API.PostgresMarshall +import Wire.Postgres +import Wire.ProposalStore (ProposalStore (..), StoredProposal (..)) + +interpretProposalStoreToPostgres :: + (PGConstraints r) => + Sem (ProposalStore ': r) a -> + Sem r a +interpretProposalStoreToPostgres = interpret $ \case + StoreProposal groupId epoch sp -> + insertProposal groupId epoch sp + GetProposal groupId epoch ref -> + selectProposal groupId epoch ref + GetAllPendingProposalRefs groupId epoch -> + selectAllPendingProposalRefs groupId epoch + GetAllPendingProposals groupId epoch -> + selectAllPendingProposals groupId epoch + DeleteAllProposals groupId -> + deleteAllProposals groupId + +insertProposal :: + (PGConstraints r) => + GroupId -> + Epoch -> + StoredProposal -> + Sem r () +insertProposal groupId epoch sp = + runStatement (groupId, epoch, sp.ref, sp.origin, sp.proposal) insert + where + insert :: + Hasql.Statement (GroupId, Epoch, ProposalRef, Maybe ProposalOrigin, RawMLS Proposal) () + insert = + lmapPG + [resultlessStatement| + INSERT INTO mls_proposal_refs (group_id, epoch, ref, origin, proposal, expires_at) + VALUES ($1 :: bytea, $2 :: int8, $3 :: bytea, $4 :: int4?, $5 :: bytea, now () + interval '28 days') + ON CONFLICT (group_id, epoch, ref) DO UPDATE + SET origin = ($4 :: int4?), + proposal = ($5 :: bytea), + expires_at = now () + interval '28 days' + |] + +selectProposal :: + (PGConstraints r) => + GroupId -> + Epoch -> + ProposalRef -> + Sem r (Maybe (RawMLS Proposal)) +selectProposal groupId epoch ref = + runStatement (groupId, epoch, ref) select + where + select :: + Hasql.Statement (GroupId, Epoch, ProposalRef) (Maybe (RawMLS Proposal)) + select = + dimapPG + [maybeStatement| + SELECT (proposal :: bytea) + FROM mls_proposal_refs + WHERE group_id = ($1 :: bytea) + AND epoch = ($2 :: int8) + AND ref = ($3 :: bytea) + AND expires_at > now () + |] + +selectAllPendingProposalRefs :: + (PGConstraints r) => + GroupId -> + Epoch -> + Sem r [ProposalRef] +selectAllPendingProposalRefs groupId epoch = + runStatement (groupId, epoch) select + where + select :: Hasql.Statement (GroupId, Epoch) [ProposalRef] + select = + dimapPG + [vectorStatement| + SELECT (ref :: bytea) + FROM mls_proposal_refs + WHERE group_id = ($1 :: bytea) + AND epoch = ($2 :: int8) + AND expires_at > now () + |] + +selectAllPendingProposals :: + (PGConstraints r) => + GroupId -> + Epoch -> + Sem r [StoredProposal] +selectAllPendingProposals groupId epoch = + fmap mkStoredProposal <$> runStatement (groupId, epoch) select + where + mkStoredProposal (ref, origin, proposal) = StoredProposal ref origin proposal + select :: + Hasql.Statement (GroupId, Epoch) [(ProposalRef, Maybe ProposalOrigin, RawMLS Proposal)] + select = + dimapPG + [vectorStatement| + SELECT (ref :: bytea), (origin :: int4?), (proposal :: bytea) + FROM mls_proposal_refs + WHERE group_id = ($1 :: bytea) + AND epoch = ($2 :: int8) + AND expires_at > now () + |] + +deleteAllProposals :: + (PGConstraints r) => + GroupId -> + Sem r () +deleteAllProposals groupId = + runStatement groupId delete + where + delete :: Hasql.Statement GroupId () + delete = + lmapPG + [resultlessStatement| + DELETE FROM mls_proposal_refs + WHERE group_id = ($1 :: bytea) + |] diff --git a/libs/wire-subsystems/wire-subsystems.cabal b/libs/wire-subsystems/wire-subsystems.cabal index 943f8fa0294..d1354b5aa0e 100644 --- a/libs/wire-subsystems/wire-subsystems.cabal +++ b/libs/wire-subsystems/wire-subsystems.cabal @@ -425,6 +425,9 @@ library Wire.PropertySubsystem.Interpreter Wire.ProposalStore Wire.ProposalStore.Cassandra + Wire.ProposalStore.DualWrite + Wire.ProposalStore.Migration + Wire.ProposalStore.Postgres Wire.RateLimit Wire.RateLimit.Interpreter Wire.Reporter diff --git a/postgres-schema.sql b/postgres-schema.sql index 948f1d62594..c06757ae0af 100644 --- a/postgres-schema.sql +++ b/postgres-schema.sql @@ -342,6 +342,22 @@ CREATE TABLE public.mls_history_client ( ALTER TABLE public.mls_history_client OWNER TO "wire-server"; +-- +-- Name: mls_proposal_refs; Type: TABLE; Schema: public; Owner: wire-server +-- + +CREATE TABLE public.mls_proposal_refs ( + group_id bytea NOT NULL, + epoch bigint NOT NULL, + ref bytea NOT NULL, + origin integer, + proposal bytea NOT NULL, + expires_at timestamp with time zone NOT NULL +); + + +ALTER TABLE public.mls_proposal_refs OWNER TO "wire-server"; + -- -- Name: remote_conversation_local_member; Type: TABLE; Schema: public; Owner: wire-server -- @@ -601,6 +617,14 @@ ALTER TABLE ONLY public.mls_history_client ADD CONSTRAINT mls_history_client_pkey PRIMARY KEY (group_id, id); +-- +-- Name: mls_proposal_refs mls_proposal_refs_pkey; Type: CONSTRAINT; Schema: public; Owner: wire-server +-- + +ALTER TABLE ONLY public.mls_proposal_refs + ADD CONSTRAINT mls_proposal_refs_pkey PRIMARY KEY (group_id, epoch, ref); + + -- -- Name: remote_conversation_local_member remote_conversation_local_member_pkey; Type: CONSTRAINT; Schema: public; Owner: wire-server -- @@ -806,6 +830,13 @@ CREATE INDEX idx_meetings_recurrence_eff_end ON public.meetings USING btree (GRE CREATE INDEX idx_meetings_start_time ON public.meetings USING btree (start_time); +-- +-- Name: mls_proposal_refs_group_epoch_expires_at_idx; Type: INDEX; Schema: public; Owner: wire-server +-- + +CREATE INDEX mls_proposal_refs_group_epoch_expires_at_idx ON public.mls_proposal_refs USING btree (group_id, epoch, expires_at); + + -- -- Name: user_group_member_user_id_idx; Type: INDEX; 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..7725acf1262 100644 --- a/services/background-worker/background-worker.integration.yaml +++ b/services/background-worker/background-worker.integration.yaml @@ -58,6 +58,7 @@ migrationOptions: migrateConversationCodes: false migrateTeamFeatures: false migrateDomainRegistration: false +migrateProposals: false # Background jobs consumer configuration for integration backgroundJobs: @@ -92,3 +93,4 @@ postgresMigration: teamFeatures: postgresql domainRegistration: postgresql user: postgresql + proposals: postgresql diff --git a/services/background-worker/src/Wire/BackgroundWorker.hs b/services/background-worker/src/Wire/BackgroundWorker.hs index b57ba12df40..29d88379c1b 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 () + cleanupProposalsMigration <- + if opts.migrateProposals + then + runAppT env $ + withNamedLogger "migrate-mls-proposal-refs" $ + Migrations.proposals 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 cleanupProposalsMigration <*> 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..ccac7845632 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, + migrateProposals :: !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..96c05bb1c9a 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.ProposalStore.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 + +proposals :: MigrationOptions -> AppT IO CleanupAction +proposals migOpts = do + cassClient <- asks (.cassandraGalley) + pgPool <- asks (.hasqlPool) + logger <- asks (.logger) + Log.info logger $ Log.msg (Log.val "starting mls proposal refs migration") + count <- register $ counter $ Prometheus.Info "wire_mls_proposal_refs_migrated_to_pg" "Number of mls proposal refs migrated to Postgresql" + finished <- register $ counter $ Prometheus.Info "wire_mls_proposal_refs_migration_finished" "Whether the mls proposal refs migration to Postgresql is finished successfully" + failed <- register $ counter $ Prometheus.Info "wire_mls_proposal_refs_migration_failed" "Whether the mls proposal refs migration to Postgresql has failed" + duration <- register $ vector "outcome" $ histogram (Prometheus.Info "wire_mls_proposal_refs_migration_duration_seconds" "Duration of mls proposal ref migration attempts") defaultBuckets + + migrationLoop <- async . lift $ migrateProposalsLoop migOpts cassClient pgPool logger count finished failed duration + + Log.info logger $ Log.msg (Log.val "started mls proposal refs migration") + pure $ do + Log.info logger $ Log.msg (Log.val "cancelling mls proposal refs 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..c0f7da0d532 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, + proposals = CassandraStorage } gundeckEndpoint = undefined brigEndpoint = undefined @@ -435,7 +436,8 @@ spec = do conversationCodes = CassandraStorage, teamFeatures = CassandraStorage, domainRegistration = CassandraStorage, - user = CassandraStorage + user = CassandraStorage, + proposals = 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..0d1b8f1e48c 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, + proposals = CassandraStorage } statuses <- newIORef mempty backendNotificationMetrics <- mkBackendNotificationMetrics diff --git a/services/brig/brig.integration.yaml b/services/brig/brig.integration.yaml index 8be11f028bd..21776a63ce1 100644 --- a/services/brig/brig.integration.yaml +++ b/services/brig/brig.integration.yaml @@ -176,6 +176,7 @@ postgresMigration: teamFeatures: postgresql domainRegistration: postgresql user: postgresql + proposals: postgresql optSettings: setActivationTimeout: 4 diff --git a/services/galley/galley.integration.yaml b/services/galley/galley.integration.yaml index 47980040c1e..e71a9a140ea 100644 --- a/services/galley/galley.integration.yaml +++ b/services/galley/galley.integration.yaml @@ -266,3 +266,4 @@ postgresMigration: teamFeatures: postgresql domainRegistration: postgresql user: postgresql + proposals: postgresql diff --git a/services/galley/src/Galley/App.hs b/services/galley/src/Galley/App.hs index bd0131b4d2d..3b1fcb3b1c7 100644 --- a/services/galley/src/Galley/App.hs +++ b/services/galley/src/Galley/App.hs @@ -150,6 +150,8 @@ import Wire.Options.Keys import Wire.ParseException import Wire.ProposalStore (ProposalStore) import Wire.ProposalStore.Cassandra +import Wire.ProposalStore.DualWrite +import Wire.ProposalStore.Postgres import Wire.RateLimit import Wire.RateLimit.Interpreter import Wire.Rpc @@ -423,6 +425,11 @@ evalGalley e = CassandraStorage -> interpretTeamFeatureStoreToCassandra MigrationToPostgresql -> interpretTeamFeatureStoreToCassandraAndPostgres PostgresqlStorage -> interpretTeamFeatureStoreToPostgres + proposalStoreInterpreter = + case (e ^. options . postgresMigration).proposals of + CassandraStorage -> interpretProposalStoreToCassandra + MigrationToPostgresql -> interpretProposalStoreToCassandraAndPostgres + PostgresqlStorage -> interpretProposalStoreToPostgres localUnit = toLocalUnsafe (e ^. options . settings . federationDomain) () teamSubsystemConfig = TeamSubsystemConfig @@ -538,7 +545,7 @@ evalGalley e = . randomToIO . runHashPassword e._options._settings._passwordHashingOptions . interpretRateLimit e._passwordHashingRateLimitEnv - . interpretProposalStoreToCassandra + . proposalStoreInterpreter . convCodesStoreInterpreter . interpretUserClientIndexStoreToCassandra (e ^. cstate) . interpretMeetingsStoreToPostgres