WPB-22955: migrate mls-proposal-refs to PostGreSQL

changelog.d/5-internal/WPB-22955

@@ -0,0 +1 @@
+Migration of mls_proposal_refs from cassandra to postgres

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 }}
 

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

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**

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: {}

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

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

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"

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'

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 ::

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);

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"

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"

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 <opensource@wire.com>
+--
+-- 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 <https://www.gnu.org/licenses/>.
+
+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