From d48eee9d10347c4bacac13212792395269741229 Mon Sep 17 00:00:00 2001 From: Gautier DI FOLCO Date: Tue, 4 Aug 2026 17:35:58 +0200 Subject: [PATCH 1/9] WPB-22954: migrate mls-commit-locks to PostGreSQL Migrate the MLS commit-lock store (galley) from Cassandra to PostgreSQL using the dual-write + background-worker pattern, following the CodeStore recipe. The Postgres acquire replicates Cassandra's IF NOT EXISTS USING TTL via INSERT ... ON CONFLICT DO UPDATE WHERE expires_at < now() RETURNING, honoring both CAS mutual exclusion and TTL expiry (a pure DO NOTHING would leave an expired lock blocking its (group_id, epoch) forever, since Postgres has no TTL reaper). Adds Postgres/DualWrite/Migration interpreters, a migration flag + metrics, helm/config/docs wiring, schema migration, and an integration test. --- changelog.d/5-internal/WPB-22954 | 1 + .../background-worker/configmap.yaml | 2 +- charts/wire-server/values.yaml | 23 +-- .../src/developer/reference/config-options.md | 104 +++-------- hack/helm_vars/common.yaml.gotmpl | 1 + hack/helm_vars/wire-server/values.yaml.gotmpl | 1 + integration/integration.cabal | 1 + .../test/Test/Migration/MLSCommitLock.hs | 74 ++++++++ .../20260804143320-mls-commit-locks.sql | 6 + .../ConversationStore/Cassandra/Queries.hs | 2 + .../src/Wire/MLSCommitLockStore/DualWrite.hs | 55 ++++++ .../src/Wire/MLSCommitLockStore/Migration.hs | 161 ++++++++++++++++++ .../src/Wire/MLSCommitLockStore/Postgres.hs | 84 +++++++++ .../src/Wire/PostgresMigrationOpts.hs | 4 +- libs/wire-subsystems/wire-subsystems.cabal | 3 + postgres-schema.sql | 21 +++ .../background-worker.integration.yaml | 1 + .../src/Wire/BackgroundWorker.hs | 11 +- .../src/Wire/BackgroundWorker/Options.hs | 2 +- .../src/Wire/PostgresMigrations.hs | 30 ++-- .../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 +- 25 files changed, 486 insertions(+), 121 deletions(-) create mode 100644 changelog.d/5-internal/WPB-22954 create mode 100644 integration/test/Test/Migration/MLSCommitLock.hs create mode 100644 libs/wire-subsystems/postgres-migrations/20260804143320-mls-commit-locks.sql create mode 100644 libs/wire-subsystems/src/Wire/MLSCommitLockStore/DualWrite.hs create mode 100644 libs/wire-subsystems/src/Wire/MLSCommitLockStore/Migration.hs create mode 100644 libs/wire-subsystems/src/Wire/MLSCommitLockStore/Postgres.hs 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 299d0703d3a..b12498bccff 100644 --- a/charts/wire-server/templates/background-worker/configmap.yaml +++ b/charts/wire-server/templates/background-worker/configmap.yaml @@ -84,7 +84,7 @@ data: migrateConversationCodes: {{ .migrateConversationCodes }} migrateTeamFeatures: {{ .migrateTeamFeatures }} migrateDomainRegistration: {{ .migrateDomainRegistration }} - migrateUsers: {{ .migrateUsers }} + migrateMLSCommitLocks: {{ .migrateMLSCommitLocks }} migrationOptions: {{ toYaml .migrationOptions | indent 6 }} diff --git a/charts/wire-server/values.yaml b/charts/wire-server/values.yaml index 9a98be5d528..e99e0611f82 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 @@ -141,8 +142,6 @@ galley: meetings: validityPeriod: "48h" - legacyTimeZone: "Europe/Berlin" - pastEditPeriod: "24h" # Optional. When set, meeting invitation emails are sent with this # sender over the configured transport (SES xor SMTP). `useSES` selects # the transport; `aws` is used when true, `smtp` when false (mirrors @@ -242,8 +241,6 @@ galley: finaliseRegardlessAfter: null # "2029-10-17T00:00:00.000Z" usersThreshold: 100 clientsThreshold: 100 - # Allow group-wise migration by clients - allowManualMigration: false lockStatus: locked limitedEventFanout: defaults: @@ -345,6 +342,10 @@ galley: defaults: status: disabled lockStatus: locked + backgroundEffects: + defaults: + status: disabled + lockStatus: locked aws: region: "eu-west-1" proxy: {} @@ -518,14 +519,6 @@ cannon: repository: quay.io/wire/nginz tag: do-not-use pullPolicy: IfNotPresent - # Image for the cannon-configurator initContainer, which only runs a - # single `echo` into a shared volume. Override repository to pull from a - # mirror registry, e.g. my-mirror.example/library/alpine - # renovate: datasource=docker depName=alpine - configuratorImage: - repository: alpine - tag: "3.24.1" - pullPolicy: IfNotPresent config: logLevel: Info logFormat: StructuredJSON @@ -1009,10 +1002,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 users - # It's important to set `settings.postgresMigration.users` to `migration-to-postgresql` + # 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. - migrateUsers: false + 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 9154c97ea73..a651f8e3011 100644 --- a/docs/src/developer/reference/config-options.md +++ b/docs/src/developer/reference/config-options.md @@ -249,28 +249,6 @@ The lock status for individual teams can be changed via the internal API (`PUT / The feature status for individual teams can be changed via the public API (if the feature is unlocked). -### Meetings validity and past-edit periods - -`settings.meetings.validityPeriod` (default `48h`) is how long a meeting stays -alive (readable and editable) after its effective end time — its `end_time`, or -the end of its recurrence window for recurring meetings; open-ended recurring -meetings never expire. `settings.meetings.pastEditPeriod` (default `24h`) bounds how -far into the past `PUT /meetings/{domain}/{id}` may move a meeting's -`start_time`/`end_time`, so past and ongoing meetings can be corrected to what -actually happened. Only provided time values are checked against this cutoff; -unchanged stored times are not re-validated — but the effective times (provided -or stored) must still satisfy `end_time > start_time`. Galley refuses to start -if `pastEditPeriod` is negative or greater than `validityPeriod`, so a meeting -edited to past times stays inside the validity window and remains visible and editable. - -```yaml -# galley.yaml -settings: - meetings: - validityPeriod: "48h" - pastEditPeriod: "24h" -``` - ### Meetings email sender and transport The optional `settings.meetings.email` block enables emailing meeting @@ -304,13 +282,6 @@ points at the path where the SMTP password is read, and path into `transport.smtpCredentials.smtpPassword`, the same pattern Brig uses for `smtp.passwordFile`. -### Meetings time settings - -The `galley.config.settings.meetings.legacyTimeZone` Helm value is an IANA time -zone id (e.g. `"Europe/Berlin"`, the default) used as the `tzid` for meetings -created by legacy clients (< V17), which send an `end_time` instead of the V17 -`duration` + `tzid` fields. It has no effect on V17+ clients. - ### Meetings Premium (deprecated) > **Deprecated (WPB-26771).** The `meetingsPremium` feature flag no longer @@ -329,20 +300,19 @@ The aggregate list endpoints (`GET /feature-configs`, `GET /teams/:tid/features`) continue to include `meetingsPremium` at all API versions, including v17. -### Background Effects (deprecated) +### Background Effects + +The `backgroundEffects` feature flag controls whether background effects are available in meetings. It is disabled and locked by default. If you want a different configuration, use the following syntax: +```yaml +backgroundEffects: + defaults: + status: disabled|enabled + lockStatus: locked|unlocked +``` -> **Deprecated (WPB-27912).** The `backgroundEffects` feature flag no longer -> affects meeting behaviour. The flag, its data type and its public/internal -> endpoints are retained for backward compatibility and are scheduled for -> removal in a future release. +The lock status for individual teams can be changed via the internal API (`PUT /i/teams/:tid/features/backgroundEffects/(un)?locked`). -The flag now defaults to **enabled and locked** and the Helm configuration -override has been removed (operators can no longer change it via Helm). The -`GET/PUT /teams/:tid/features/backgroundEffects` and internal lock-status -endpoints return 404 at API version v17; they remain available through v16. -The aggregate endpoints `GET /feature-configs` and -`GET /teams/:tid/features` continue to include `backgroundEffects` at all API -versions, including v17. +The feature status for individual teams can be changed via the public API (if the feature is unlocked). ### File Sharing @@ -392,21 +362,6 @@ The settings mean: - `deletionTimeoutDuration`: how long to keep an adminless conversation before it is deleted. - `reminderTimeoutDurations`: when before deletion reminder notifications should be sent. -In federated conversations, automatic senderless deletion is skipped when the -conversation contains remote members because the corresponding system delete -event cannot yet be sent safely to the remote backend. This applies both when -the feature is enabled and existing conversations are scanned without an -origin user, and when a previously scheduled senderless deletion job runs. -Reminders for a skipped deletion are also skipped because they would be -misleading. The skipped deletion is logged at info level. - -Autopromotion still runs because the conversation-owning backend stores the -authoritative member roles. Remote clients may miss the immediate senderless -member-update notification, but a subsequent conversation fetch obtains the -current role from the owning backend. Member updates and deletions with an -origin user continue to use the existing ordinary federation events and are -not skipped. - Durations are strings with a number and a unit suffix. Supported units are `us`, `ms`, `s`, `m`, `h`, `d`, and `w`. It is **not** recommended or supported to set these below a day in production environments. Feature responses, including `GET /feature-configs`, `GET /teams/:tid/features`, and `GET /teams/:tid/features/preventAdminlessGroups`, include the duration fields: @@ -425,7 +380,7 @@ Feature responses, including `GET /feature-configs`, `GET /teams/:tid/features`, From a client's perspective, API versioning works like this: -- API version V18 and newer should send the duration fields to `PUT /teams/:tid/features/preventAdminlessGroups`. +- API version V17 and newer should send the duration fields to `PUT /teams/:tid/features/preventAdminlessGroups`. - Feature responses include the duration fields for clients to read. The lock status for individual teams can be changed via the internal API (`PUT /i/teams/:tid/features/preventAdminlessGroups/(un)?locked`). @@ -2092,12 +2047,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 @@ -2128,7 +2085,7 @@ The current settings and their background-worker flags are: - `conversationCodes` -> `migrateConversationCodes` - `teamFeatures` -> `migrateTeamFeatures` - `domainRegistration` -> `migrateDomainRegistration` -- `user` -> `migrateUsers` +- `mlsCommitLocks` -> `migrateMLSCommitLocks` **Migration pattern per migration setting** @@ -2147,15 +2104,15 @@ The current settings and their background-worker flags are: conversation: migration-to-postgresql conversationCodes: migration-to-postgresql teamFeatures: migration-to-postgresql - domainRegistration: migration-to-postgresql - user: migration-to-postgresql + domainRegistration: cassandra + mlsCommitLocks: cassandra background-worker: config: migrateConversations: false migrateConversationCodes: false migrateTeamFeatures: false - migrateDomainRegistration: false - migrateUsers: false + migrateDomainRegistration: false + migrateMLSCommitLocks: false ``` This change should restart the affected pods, and new writes will follow the @@ -2169,8 +2126,8 @@ The current settings and their background-worker flags are: migrateConversations: true migrateConversationCodes: true migrateTeamFeatures: true - migrateDomainRegistration: true - migrateUsers: true + migrateDomainRegistration: true + migrateMLSCommitLocks: true ``` During migration, Cassandra rows are not deleted. Writes and migration share @@ -2186,16 +2143,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` - - `user`: `wire_user_migration_finished` - - > ⚠️ For user migrations please watch the logs for `Invalid user found, - > skipping`. This would be accompanied by an error which is either - > `UserHasNoName` or `UserHasNoActivated`. These users are invalid and all - > interactions with them were resulting in errors. If these warnings are - > ignored, these users will stop existing in the system. If these users are - > to be saved, the operator must insert some value as `name` and/or - > `activated` and then re-trigger the migration **after** the background - > worker finishes migrating the valid users. + - `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,15 +2156,15 @@ The current settings and their background-worker flags are: conversation: postgresql conversationCodes: postgresql teamFeatures: postgresql - domainRegistration: postgresql - user: postgresql + domainRegistration: cassandra + mlsCommitLocks: cassandra background-worker: config: migrateConversations: false migrateConversationCodes: false migrateTeamFeatures: false - migrateDomainRegistration: false - migrateUsers: 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 43373b1cf2e..80ca1d0d95a 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 8dd9466197c..622b4fbb087 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.User Test.Migration.Util 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 66662926e3d..48c6a157a5b 100644 --- a/libs/wire-subsystems/wire-subsystems.cabal +++ b/libs/wire-subsystems/wire-subsystems.cabal @@ -404,6 +404,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 2d3df5fb27f..4ea7d87802c 100644 --- a/postgres-schema.sql +++ b/postgres-schema.sql @@ -1481,6 +1481,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 -- @@ -1966,6 +1979,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 b0bd0d172e4..4ef831db7de 100644 --- a/services/background-worker/background-worker.integration.yaml +++ b/services/background-worker/background-worker.integration.yaml @@ -93,3 +93,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 6c12b02e816..937ee8b4575 100644 --- a/services/background-worker/src/Wire/BackgroundWorker.hs +++ b/services/background-worker/src/Wire/BackgroundWorker.hs @@ -78,14 +78,13 @@ run opts galleyOpts = do withNamedLogger "migrate-domain-registration" $ Migrations.domainRegistration opts.migrationOptions else pure $ pure () - cleanupUsersMigration <- - if opts.migrateUsers + cleanupMLSLocksMigration <- + if opts.migrateMLSCommitLocks then runAppT env $ - withNamedLogger "migrate-users" $ - Migrations.users opts.migrationOptions + withNamedLogger "migrate-mls-commit-locks" $ + Migrations.mlsCommitLocks opts.migrationOptions else pure $ pure () - cleanupJobs <- runAppT env $ withNamedLogger "background-job-consumer" $ @@ -104,7 +103,7 @@ run opts galleyOpts = do <*> Concurrently cleanUpConvCodesMigration <*> Concurrently cleanupTeamFeaturesMigration <*> Concurrently cleanupDomainRegistrationMigration - <*> Concurrently cleanupUsersMigration + <*> 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 035460cfc32..f75fcbb8b35 100644 --- a/services/background-worker/src/Wire/BackgroundWorker/Options.hs +++ b/services/background-worker/src/Wire/BackgroundWorker/Options.hs @@ -55,7 +55,7 @@ data Opts = Opts migrateConversationCodes :: !Bool, migrateTeamFeatures :: !Bool, migrateDomainRegistration :: !Bool, - migrateUsers :: !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 28c6a789a4a..3c3f64fdc1e 100644 --- a/services/background-worker/src/Wire/PostgresMigrations.hs +++ b/services/background-worker/src/Wire/PostgresMigrations.hs @@ -24,11 +24,11 @@ import UnliftIO import Wire.BackgroundWorker.Env import Wire.BackgroundWorker.Util import Wire.CodeStore.Migration -import Wire.ConversationStore.Migration qualified as ConversationStore +import Wire.ConversationStore.Migration import Wire.DomainRegistrationStore.Migration +import Wire.MLSCommitLockStore.Migration import Wire.Migration (MigrationOptions) import Wire.TeamFeatureStore.Migration -import Wire.UserStore.Migration qualified as UserStore conversations :: MigrationOptions -> AppT IO CleanupAction conversations migOpts = do @@ -46,8 +46,8 @@ conversations migOpts = do userMigFailed <- register $ counter $ Prometheus.Info "wire_user_remote_convs_migration_failed" "Whether the migration of remote conversation membership data to Postgresql has failed" userMigDuration <- register $ vector "outcome" $ histogram (Prometheus.Info "wire_user_remote_convs_migration_duration_seconds" "Duration of remote conversation membership migration attempts") defaultBuckets - convLoop <- async . lift $ ConversationStore.migrateConvsLoop migOpts cassClient pgPool logger convMigCounter convMigFinished convMigFailed convMigDuration - userLoop <- async . lift $ ConversationStore.migrateUsersLoop migOpts cassClient pgPool logger userMigCounter userMigFinished userMigFailed userMigDuration + convLoop <- async . lift $ migrateConvsLoop migOpts cassClient pgPool logger convMigCounter convMigFinished convMigFailed convMigDuration + userLoop <- async . lift $ migrateUsersLoop migOpts cassClient pgPool logger userMigCounter userMigFinished userMigFailed userMigDuration Log.info logger $ Log.msg (Log.val "started conversation migration") pure $ do @@ -109,20 +109,20 @@ domainRegistration migOpts = do Log.info logger $ Log.msg (Log.val "cancelling domain registration migration") cancel migrationLoop -users :: MigrationOptions -> AppT IO CleanupAction -users migOpts = do - cassClient <- asks (.cassandraBrig) +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 user migration") - count <- register $ counter $ Prometheus.Info "wire_users_migrated_to_pg" "Number of user rows migrated to Postgresql" - finished <- register $ counter $ Prometheus.Info "wire_users_migration_finished" "Whether the user migration to Postgresql is finished successfully" - failed <- register $ counter $ Prometheus.Info "wire_users_migration_failed" "Whether the user migration to Postgresql has failed" - duration <- register $ vector "outcome" $ histogram (Prometheus.Info "wire_users_migration_duration_seconds" "Duration of user migration attempts") defaultBuckets + 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 $ UserStore.migrateUsersLoop migOpts cassClient pgPool logger count finished failed duration + migrationLoop <- async . lift $ migrateMLSCommitLocksLoop migOpts cassClient pgPool logger count finished failed duration - Log.info logger $ Log.msg (Log.val "started user migration") + Log.info logger $ Log.msg (Log.val "started mls commit locks migration") pure $ do - Log.info logger $ Log.msg (Log.val "cancelling user migration") + 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 be46a03c648..41f72bb1dd2 100644 --- a/services/background-worker/test/Test/Wire/BackendNotificationPusherSpec.hs +++ b/services/background-worker/test/Test/Wire/BackendNotificationPusherSpec.hs @@ -496,7 +496,8 @@ spec = do conversationCodes = CassandraStorage, teamFeatures = CassandraStorage, domainRegistration = CassandraStorage, - user = CassandraStorage + user = CassandraStorage, + mlsCommitLocks = CassandraStorage } gundeckEndpoint = undefined brigEndpoint = undefined @@ -560,7 +561,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 34762762965..25562145d79 100644 --- a/services/galley/galley.integration.yaml +++ b/services/galley/galley.integration.yaml @@ -268,3 +268,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 4d755d8c612..724e91f5536 100644 --- a/services/galley/src/Galley/App.hs +++ b/services/galley/src/Galley/App.hs @@ -109,6 +109,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 @@ -440,6 +442,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 @@ -542,7 +549,7 @@ evalGalley e = . interpretTeamMemberStoreToCassandraWithPaging lh . interpretTeamMemberStoreToCassandra lh . teamFeatureStoreInterpreter - . interpretMLSCommitLockStoreToCassandra (e ^. cstate) + . mlsCommitLockStoreInterpreter . convStoreInterpreter . interpretTeamNotificationStoreToCassandra . interpretServiceStoreToCassandra (e ^. cstate) From ea9c7eb70987e81f43cf9e4c7862840f2049b579 Mon Sep 17 00:00:00 2001 From: Gautier DI FOLCO Date: Thu, 3 Sep 2026 02:25:00 +0200 Subject: [PATCH 2/9] WPB-22954: use pg advisory locks for mls commit locks Replaces the mls_commit_locks table + dual-write + backfill migration from the previous commit with a direct cutover to session-scoped pg advisory locks (Wire.MigrationLock try-lock). Contention answers stale-message immediately; crash safety comes from the holding connection dying instead of the 10-minute TTL. Deletes the MLSCommitLockStore.DualWrite/Migration modules, the background-worker migration job, the 20260804143320 SQL migration and all mlsCommitLocks/migrateMLSCommitLocks config knobs. The cassandra mls_commit_locks table becomes unread; dropping it (and galley schema V68) is a follow-up. --- changelog.d/5-internal/WPB-22954 | 4 +- .../background-worker/configmap.yaml | 1 - charts/wire-server/values.yaml | 5 - .../src/developer/reference/config-options.md | 9 - hack/helm_vars/common.yaml.gotmpl | 1 - hack/helm_vars/wire-server/values.yaml.gotmpl | 1 - integration/integration.cabal | 2 +- integration/test/Test/MLS/CommitLock.hs | 40 +++++ .../test/Test/Migration/MLSCommitLock.hs | 74 -------- .../20260804143320-mls-commit-locks.sql | 6 - .../src/Wire/ConversationStore.hs | 11 +- .../src/Wire/ConversationStore/Cassandra.hs | 43 +---- .../ConversationStore/Cassandra/Queries.hs | 8 - .../ConversationSubsystem/Action/Reset.hs | 2 - .../Wire/ConversationSubsystem/Federation.hs | 3 - .../MLS/Commit/ExternalCommit.hs | 2 - .../MLS/Commit/InternalCommit.hs | 2 - .../Wire/ConversationSubsystem/MLS/Message.hs | 4 - .../MLS/SubConversation.hs | 5 - .../Wire/ConversationSubsystem/MLS/Util.hs | 62 +++---- .../src/Wire/MLSCommitLockStore/DualWrite.hs | 55 ------ .../src/Wire/MLSCommitLockStore/Migration.hs | 161 ------------------ .../src/Wire/MLSCommitLockStore/Postgres.hs | 103 ++++++----- .../wire-subsystems/src/Wire/MigrationLock.hs | 118 +++++++++++++ .../src/Wire/PostgresMigrationOpts.hs | 4 +- libs/wire-subsystems/wire-subsystems.cabal | 2 - postgres-schema.sql | 22 --- .../background-worker.integration.yaml | 1 - .../src/Wire/BackgroundWorker.hs | 10 +- .../src/Wire/BackgroundWorker/Options.hs | 1 - .../background-worker/src/Wire/Effects.hs | 5 +- .../src/Wire/PostgresMigrations.hs | 18 -- .../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 | 10 +- services/galley/test/integration/API/MLS.hs | 43 ----- 38 files changed, 256 insertions(+), 593 deletions(-) create mode 100644 integration/test/Test/MLS/CommitLock.hs delete mode 100644 integration/test/Test/Migration/MLSCommitLock.hs delete mode 100644 libs/wire-subsystems/postgres-migrations/20260804143320-mls-commit-locks.sql delete mode 100644 libs/wire-subsystems/src/Wire/MLSCommitLockStore/DualWrite.hs delete mode 100644 libs/wire-subsystems/src/Wire/MLSCommitLockStore/Migration.hs diff --git a/changelog.d/5-internal/WPB-22954 b/changelog.d/5-internal/WPB-22954 index 133c43a40bd..ecc0777a956 100644 --- a/changelog.d/5-internal/WPB-22954 +++ b/changelog.d/5-internal/WPB-22954 @@ -1 +1,3 @@ -Migration of mls commit locks from cassandra to postgres +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/charts/wire-server/templates/background-worker/configmap.yaml b/charts/wire-server/templates/background-worker/configmap.yaml index b12498bccff..d4fe2a63202 100644 --- a/charts/wire-server/templates/background-worker/configmap.yaml +++ b/charts/wire-server/templates/background-worker/configmap.yaml @@ -84,7 +84,6 @@ 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 e99e0611f82..f6448ead810 100644 --- a/charts/wire-server/values.yaml +++ b/charts/wire-server/values.yaml @@ -90,7 +90,6 @@ galley: teamFeatures: cassandra domainRegistration: cassandra user: cassandra - mlsCommitLocks: cassandra settings: httpPoolSize: 128 maxTeamSize: 10000 @@ -1002,10 +1001,6 @@ 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 a651f8e3011..fa89f36beca 100644 --- a/docs/src/developer/reference/config-options.md +++ b/docs/src/developer/reference/config-options.md @@ -2047,14 +2047,12 @@ 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 @@ -2085,7 +2083,6 @@ The current settings and their background-worker flags are: - `conversationCodes` -> `migrateConversationCodes` - `teamFeatures` -> `migrateTeamFeatures` - `domainRegistration` -> `migrateDomainRegistration` -- `mlsCommitLocks` -> `migrateMLSCommitLocks` **Migration pattern per migration setting** @@ -2105,14 +2102,12 @@ The current settings and their background-worker flags are: conversationCodes: migration-to-postgresql teamFeatures: migration-to-postgresql domainRegistration: cassandra - mlsCommitLocks: cassandra background-worker: config: migrateConversations: false migrateConversationCodes: false migrateTeamFeatures: false migrateDomainRegistration: false - migrateMLSCommitLocks: false ``` This change should restart the affected pods, and new writes will follow the @@ -2127,7 +2122,6 @@ The current settings and their background-worker flags are: migrateConversationCodes: true migrateTeamFeatures: true migrateDomainRegistration: true - migrateMLSCommitLocks: true ``` During migration, Cassandra rows are not deleted. Writes and migration share @@ -2143,7 +2137,6 @@ 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 @@ -2157,14 +2150,12 @@ The current settings and their background-worker flags are: conversationCodes: postgresql teamFeatures: postgresql domainRegistration: cassandra - mlsCommitLocks: cassandra background-worker: config: migrateConversations: false migrateConversationCodes: false migrateTeamFeatures: 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 313b4f339bf..2276355e2a9 100644 --- a/hack/helm_vars/common.yaml.gotmpl +++ b/hack/helm_vars/common.yaml.gotmpl @@ -19,7 +19,6 @@ 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 80ca1d0d95a..43373b1cf2e 100644 --- a/hack/helm_vars/wire-server/values.yaml.gotmpl +++ b/hack/helm_vars/wire-server/values.yaml.gotmpl @@ -306,7 +306,6 @@ 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 622b4fbb087..2195b8d0433 100644 --- a/integration/integration.cabal +++ b/integration/integration.cabal @@ -180,12 +180,12 @@ library Test.Migration.Conversation Test.Migration.ConversationCodes Test.Migration.DomainRegistration - Test.Migration.MLSCommitLock Test.Migration.TeamFeatures Test.Migration.User 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..3b34ff14f9b --- /dev/null +++ b/integration/test/Test/MLS/CommitLock.hs @@ -0,0 +1,40 @@ +-- 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 <- randomUser OwnDomain def + alice1 <- createMLSClient def alice + bob <- randomUser OwnDomain def + bob1 <- createMLSClient def bob + void $ uploadNewKeyPackage def bob1 + convId <- createNewGroup def alice1 + void $ createAddCommit alice1 convId [bob] >>= sendAndConsumeCommitBundle + + charlie <- randomUser OwnDomain def + charlie1 <- createMLSClient def charlie + void $ uploadNewKeyPackage def charlie1 + void $ createAddCommit alice1 convId [charlie] >>= sendAndConsumeCommitBundle diff --git a/integration/test/Test/Migration/MLSCommitLock.hs b/integration/test/Test/Migration/MLSCommitLock.hs deleted file mode 100644 index 7c3ccaef23c..00000000000 --- a/integration/test/Test/Migration/MLSCommitLock.hs +++ /dev/null @@ -1,74 +0,0 @@ --- 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 deleted file mode 100644 index 58d901b8047..00000000000 --- a/libs/wire-subsystems/postgres-migrations/20260804143320-mls-commit-locks.sql +++ /dev/null @@ -1,6 +0,0 @@ -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.hs b/libs/wire-subsystems/src/Wire/ConversationStore.hs index a8715073b8f..4bb949a2e99 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,12 @@ 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..74ac0567634 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,36 +349,7 @@ 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 @@ -880,14 +849,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. diff --git a/libs/wire-subsystems/src/Wire/ConversationStore/Cassandra/Queries.hs b/libs/wire-subsystems/src/Wire/ConversationStore/Cassandra/Queries.hs index 11cf1afb8d6..7b8a7a466cb 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,13 +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 = ?" -selectAllCommitLocks :: PrepQuery R () (GroupId, Epoch) -selectAllCommitLocks = "select group_id, epoch from mls_commit_locks" -- Bots --------------------------------------------------------------------- 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/DualWrite.hs b/libs/wire-subsystems/src/Wire/MLSCommitLockStore/DualWrite.hs deleted file mode 100644 index 21d916af3fb..00000000000 --- a/libs/wire-subsystems/src/Wire/MLSCommitLockStore/DualWrite.hs +++ /dev/null @@ -1,55 +0,0 @@ --- 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 deleted file mode 100644 index 0e14b4d8dbb..00000000000 --- a/libs/wire-subsystems/src/Wire/MLSCommitLockStore/Migration.hs +++ /dev/null @@ -1,161 +0,0 @@ --- 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 index dbc627812f9..ef65b338729 100644 --- a/libs/wire-subsystems/src/Wire/MLSCommitLockStore/Postgres.hs +++ b/libs/wire-subsystems/src/Wire/MLSCommitLockStore/Postgres.hs @@ -1,3 +1,4 @@ +{-# OPTIONS_GHC -Wno-orphans #-} -- This file is part of the Wire Server implementation. -- -- Copyright (C) 2026 Wire Swiss GmbH @@ -20,65 +21,61 @@ module Wire.MLSCommitLockStore.Postgres ) where -import Hasql.Statement qualified as Hasql -import Hasql.TH +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 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) +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'. -- --- Acquire replicates Cassandra's @INSERT ... IF NOT EXISTS USING TTL@ as an --- @INSERT ... ON CONFLICT DO UPDATE ... WHERE expires_at < now() RETURNING@: +-- 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. -- --- * 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. +-- 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) => + ( PGConstraints r, + Member Resource r, + Member TinyLog 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 - |] +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 -releaseStmt :: Hasql.Statement (GroupId, Epoch) () -releaseStmt = - lmapPG - [resultlessStatement| - DELETE FROM mls_commit_locks - WHERE group_id = ($1 :: bytea) AND epoch = ($2 :: int8) - |] +-- | 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..63059d7693d 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.Exception (onException) +import Control.Concurrent.Async qualified as Async 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/src/Wire/PostgresMigrationOpts.hs b/libs/wire-subsystems/src/Wire/PostgresMigrationOpts.hs index fa12ac9e8bc..327862f7cd5 100644 --- a/libs/wire-subsystems/src/Wire/PostgresMigrationOpts.hs +++ b/libs/wire-subsystems/src/Wire/PostgresMigrationOpts.hs @@ -56,8 +56,7 @@ data PostgresMigrationOpts = PostgresMigrationOpts conversationCodes :: StorageLocation, teamFeatures :: StorageLocation, domainRegistration :: StorageLocation, - user :: StorageLocation, - mlsCommitLocks :: StorageLocation + user :: StorageLocation } deriving (Show) @@ -69,4 +68,3 @@ 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 48c6a157a5b..efb31a6feff 100644 --- a/libs/wire-subsystems/wire-subsystems.cabal +++ b/libs/wire-subsystems/wire-subsystems.cabal @@ -404,8 +404,6 @@ library Wire.MeetingsSubsystem.Notification Wire.Migration Wire.MigrationLock - Wire.MLSCommitLockStore.DualWrite - Wire.MLSCommitLockStore.Migration Wire.MLSCommitLockStore.Postgres Wire.MlsKeyPackageStore Wire.MlsKeyPackageStore.Cassandra diff --git a/postgres-schema.sql b/postgres-schema.sql index 4ea7d87802c..c96cc12aa37 100644 --- a/postgres-schema.sql +++ b/postgres-schema.sql @@ -1481,20 +1481,6 @@ 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 -- @@ -1979,14 +1965,6 @@ 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 4ef831db7de..b0bd0d172e4 100644 --- a/services/background-worker/background-worker.integration.yaml +++ b/services/background-worker/background-worker.integration.yaml @@ -93,4 +93,3 @@ 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 937ee8b4575..b57ba12df40 100644 --- a/services/background-worker/src/Wire/BackgroundWorker.hs +++ b/services/background-worker/src/Wire/BackgroundWorker.hs @@ -78,13 +78,6 @@ 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" $ @@ -96,14 +89,13 @@ 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 f75fcbb8b35..61df5d5d14f 100644 --- a/services/background-worker/src/Wire/BackgroundWorker/Options.hs +++ b/services/background-worker/src/Wire/BackgroundWorker/Options.hs @@ -55,7 +55,6 @@ 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/Effects.hs b/services/background-worker/src/Wire/Effects.hs index 0d367f19eed..c45c162ae17 100644 --- a/services/background-worker/src/Wire/Effects.hs +++ b/services/background-worker/src/Wire/Effects.hs @@ -73,7 +73,8 @@ 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.MLSCommitLockStore.Postgres (interpretMLSCommitLockStoreToPostgres) import Wire.ConversationSubsystem (ConversationSubsystem) import Wire.ConversationSubsystem.Interpreter (ConversationSubsystemError, GroupInfoCheckEnabled (..), IntraListing (..), interpretConversationSubsystem) import Wire.ExternalAccess (ExternalAccess) @@ -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/background-worker/src/Wire/PostgresMigrations.hs b/services/background-worker/src/Wire/PostgresMigrations.hs index 3c3f64fdc1e..e837495f591 100644 --- a/services/background-worker/src/Wire/PostgresMigrations.hs +++ b/services/background-worker/src/Wire/PostgresMigrations.hs @@ -26,7 +26,6 @@ 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 @@ -109,20 +108,3 @@ domainRegistration migOpts = 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 41f72bb1dd2..be46a03c648 100644 --- a/services/background-worker/test/Test/Wire/BackendNotificationPusherSpec.hs +++ b/services/background-worker/test/Test/Wire/BackendNotificationPusherSpec.hs @@ -496,8 +496,7 @@ spec = do conversationCodes = CassandraStorage, teamFeatures = CassandraStorage, domainRegistration = CassandraStorage, - user = CassandraStorage, - mlsCommitLocks = CassandraStorage + user = CassandraStorage } gundeckEndpoint = undefined brigEndpoint = undefined @@ -561,8 +560,7 @@ spec = do conversationCodes = CassandraStorage, teamFeatures = CassandraStorage, domainRegistration = CassandraStorage, - user = CassandraStorage, - mlsCommitLocks = CassandraStorage + user = 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 f4511334905..5d89532bfec 100644 --- a/services/background-worker/test/Test/Wire/Util.hs +++ b/services/background-worker/test/Test/Wire/Util.hs @@ -50,8 +50,7 @@ testEnv = do conversationCodes = CassandraStorage, teamFeatures = CassandraStorage, domainRegistration = CassandraStorage, - user = CassandraStorage, - mlsCommitLocks = CassandraStorage + user = CassandraStorage } statuses <- newIORef mempty backendNotificationMetrics <- mkBackendNotificationMetrics diff --git a/services/brig/brig.integration.yaml b/services/brig/brig.integration.yaml index c428d24068a..8be11f028bd 100644 --- a/services/brig/brig.integration.yaml +++ b/services/brig/brig.integration.yaml @@ -176,7 +176,6 @@ 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 25562145d79..34762762965 100644 --- a/services/galley/galley.integration.yaml +++ b/services/galley/galley.integration.yaml @@ -268,4 +268,3 @@ 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 724e91f5536..d99a52c814e 100644 --- a/services/galley/src/Galley/App.hs +++ b/services/galley/src/Galley/App.hs @@ -108,8 +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.MLSCommitLockStore.DualWrite (interpretMLSCommitLockStoreToCassandraAndPostgres) +import Wire.ConversationStore.Cassandra (MigrationError (..), interpretConversationStoreByMigration) import Wire.MLSCommitLockStore.Postgres (interpretMLSCommitLockStoreToPostgres) import Wire.ConversationSubsystem import Wire.ConversationSubsystem.Interpreter (ConversationSubsystemError, GroupInfoCheckEnabled (..), IntraListing (IntraListing), interpretConversationSubsystem) @@ -442,11 +441,6 @@ 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 @@ -549,7 +543,7 @@ evalGalley e = . interpretTeamMemberStoreToCassandraWithPaging lh . interpretTeamMemberStoreToCassandra lh . teamFeatureStoreInterpreter - . mlsCommitLockStoreInterpreter + . 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) From 6c33c1fccbed6a92019aede036631f79e1b0651b Mon Sep 17 00:00:00 2001 From: Gautier DI FOLCO Date: Thu, 3 Sep 2026 03:27:43 +0200 Subject: [PATCH 3/9] WPB-22954: apply formatter (stanza import order, ormolu) --- libs/wire-subsystems/src/Wire/ConversationStore.hs | 1 - libs/wire-subsystems/src/Wire/ConversationStore/Cassandra.hs | 3 --- .../src/Wire/ConversationStore/Cassandra/Queries.hs | 1 - libs/wire-subsystems/src/Wire/MLSCommitLockStore/Postgres.hs | 4 +++- libs/wire-subsystems/src/Wire/MigrationLock.hs | 2 +- services/background-worker/src/Wire/Effects.hs | 2 +- services/background-worker/src/Wire/PostgresMigrations.hs | 1 - services/galley/src/Galley/App.hs | 2 +- 8 files changed, 6 insertions(+), 10 deletions(-) diff --git a/libs/wire-subsystems/src/Wire/ConversationStore.hs b/libs/wire-subsystems/src/Wire/ConversationStore.hs index 4bb949a2e99..de97b4fe10a 100644 --- a/libs/wire-subsystems/src/Wire/ConversationStore.hs +++ b/libs/wire-subsystems/src/Wire/ConversationStore.hs @@ -43,7 +43,6 @@ import Wire.Sem.Paging.Cassandra import Wire.StoredConversation import Wire.UserList - data MLSCommitLockStore m a where -- | Runs the action while holding an exclusive lock for @(groupId, epoch)@. -- Returns 'Nothing' without running the action when another holder is active diff --git a/libs/wire-subsystems/src/Wire/ConversationStore/Cassandra.hs b/libs/wire-subsystems/src/Wire/ConversationStore/Cassandra.hs index 74ac0567634..dd6514412a0 100644 --- a/libs/wire-subsystems/src/Wire/ConversationStore/Cassandra.hs +++ b/libs/wire-subsystems/src/Wire/ConversationStore/Cassandra.hs @@ -349,8 +349,6 @@ updateToMLSProtocol client cnv = updateChannelAddPermissions :: ConvId -> AddPermission -> Client () updateChannelAddPermissions cid cap = retry x5 $ write Cql.updateChannelAddPermission (params LocalQuorum (cap, cid)) - - removeTeamConv :: TeamId -> ConvId -> Client () removeTeamConv tid cid = liftClient $ do retry x5 . batch $ do @@ -849,7 +847,6 @@ isConversationOutOfSync cid = maybe False (fromMaybe False . runIdentity) <$> retry x1 (query1 Cql.lookupConvOutOfSync (params LocalQuorum (Identity cid))) - 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 7b8a7a466cb..a75a144bcf9 100644 --- a/libs/wire-subsystems/src/Wire/ConversationStore/Cassandra/Queries.hs +++ b/libs/wire-subsystems/src/Wire/ConversationStore/Cassandra/Queries.hs @@ -354,7 +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 = ?" - -- Bots --------------------------------------------------------------------- insertBot :: PrepQuery W (ConvId, BotId, ServiceId, ProviderId) () diff --git a/libs/wire-subsystems/src/Wire/MLSCommitLockStore/Postgres.hs b/libs/wire-subsystems/src/Wire/MLSCommitLockStore/Postgres.hs index ef65b338729..70512f5aff1 100644 --- a/libs/wire-subsystems/src/Wire/MLSCommitLockStore/Postgres.hs +++ b/libs/wire-subsystems/src/Wire/MLSCommitLockStore/Postgres.hs @@ -1,4 +1,5 @@ {-# OPTIONS_GHC -Wno-orphans #-} + -- This file is part of the Wire Server implementation. -- -- Copyright (C) 2026 Wire Swiss GmbH @@ -76,6 +77,7 @@ instance MigrationLockable (GroupId, Epoch) where (fromIntegral (hash (unGroupId gId)) :: Int64) `rotateL` 31 `xor` fromIntegral (epochNumber epoch) toText (gId, epoch) = - "0x" <> decodeUtf8 (hex (unGroupId gId)) + "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 63059d7693d..6b0869dd919 100644 --- a/libs/wire-subsystems/src/Wire/MigrationLock.hs +++ b/libs/wire-subsystems/src/Wire/MigrationLock.hs @@ -36,8 +36,8 @@ module Wire.MigrationLock where -import Control.Exception (onException) import Control.Concurrent.Async qualified as Async +import Control.Exception (onException) import Data.Bits import Data.Hashable (hash) import Data.Id diff --git a/services/background-worker/src/Wire/Effects.hs b/services/background-worker/src/Wire/Effects.hs index c45c162ae17..6988c406629 100644 --- a/services/background-worker/src/Wire/Effects.hs +++ b/services/background-worker/src/Wire/Effects.hs @@ -74,7 +74,6 @@ import Wire.CodeStore.DualWrite (interpretCodeStoreToCassandraAndPostgres) import Wire.CodeStore.Postgres (interpretCodeStoreToPostgres) import Wire.ConversationStore (ConversationStore, MLSCommitLockStore) import Wire.ConversationStore.Cassandra (MigrationError (..), interpretConversationStoreByMigration) -import Wire.MLSCommitLockStore.Postgres (interpretMLSCommitLockStoreToPostgres) import Wire.ConversationSubsystem (ConversationSubsystem) import Wire.ConversationSubsystem.Interpreter (ConversationSubsystemError, GroupInfoCheckEnabled (..), IntraListing (..), interpretConversationSubsystem) import Wire.ExternalAccess (ExternalAccess) @@ -97,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) diff --git a/services/background-worker/src/Wire/PostgresMigrations.hs b/services/background-worker/src/Wire/PostgresMigrations.hs index e837495f591..604cab0140c 100644 --- a/services/background-worker/src/Wire/PostgresMigrations.hs +++ b/services/background-worker/src/Wire/PostgresMigrations.hs @@ -107,4 +107,3 @@ domainRegistration migOpts = do pure $ do Log.info logger $ Log.msg (Log.val "cancelling domain registration migration") cancel migrationLoop - diff --git a/services/galley/src/Galley/App.hs b/services/galley/src/Galley/App.hs index d99a52c814e..62e99f4a40b 100644 --- a/services/galley/src/Galley/App.hs +++ b/services/galley/src/Galley/App.hs @@ -109,7 +109,6 @@ import Wire.CodeStore.DualWrite import Wire.CodeStore.Postgres import Wire.ConversationStore (ConversationStore, MLSCommitLockStore) import Wire.ConversationStore.Cassandra (MigrationError (..), interpretConversationStoreByMigration) -import Wire.MLSCommitLockStore.Postgres (interpretMLSCommitLockStoreToPostgres) import Wire.ConversationSubsystem import Wire.ConversationSubsystem.Interpreter (ConversationSubsystemError, GroupInfoCheckEnabled (..), IntraListing (IntraListing), interpretConversationSubsystem) import Wire.CustomBackendStore @@ -138,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) From cee5cd95cce5addb341808324a036524a9ff2b3a Mon Sep 17 00:00:00 2001 From: Gautier DI FOLCO Date: Thu, 3 Sep 2026 08:34:41 +0200 Subject: [PATCH 4/9] Hello CI From 8088f8b4fd2448cf6355c3fc462b52bdd3670221 Mon Sep 17 00:00:00 2001 From: Gautier DI FOLCO Date: Thu, 3 Sep 2026 11:11:21 +0200 Subject: [PATCH 5/9] WPB-22954: restore upstream content lost in rebase A rebase onto develop with bad conflict resolutions silently reverted newer upstream work: cannon.configuratorImage in the wire-server chart values (broke the kube-integration CI helm render at cannon/statefulset.yaml:146), the background-worker user-migration wiring (migrateUsers), meetings time-setting values/docs, and a separator line in postgres-schema.sql. Restore all of these from origin/develop; the MLS advisory-lock cutover needs none of the reverted content. --- .../background-worker/configmap.yaml | 1 + charts/wire-server/values.yaml | 20 +++- .../src/developer/reference/config-options.md | 95 +++++++++++++++---- postgres-schema.sql | 1 + .../src/Wire/BackgroundWorker.hs | 11 ++- .../src/Wire/BackgroundWorker/Options.hs | 1 + .../src/Wire/PostgresMigrations.hs | 25 ++++- 7 files changed, 129 insertions(+), 25 deletions(-) diff --git a/charts/wire-server/templates/background-worker/configmap.yaml b/charts/wire-server/templates/background-worker/configmap.yaml index d4fe2a63202..299d0703d3a 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 }} + migrateUsers: {{ .migrateUsers }} migrationOptions: {{ toYaml .migrationOptions | indent 6 }} diff --git a/charts/wire-server/values.yaml b/charts/wire-server/values.yaml index f6448ead810..9a98be5d528 100644 --- a/charts/wire-server/values.yaml +++ b/charts/wire-server/values.yaml @@ -141,6 +141,8 @@ galley: meetings: validityPeriod: "48h" + legacyTimeZone: "Europe/Berlin" + pastEditPeriod: "24h" # Optional. When set, meeting invitation emails are sent with this # sender over the configured transport (SES xor SMTP). `useSES` selects # the transport; `aws` is used when true, `smtp` when false (mirrors @@ -240,6 +242,8 @@ galley: finaliseRegardlessAfter: null # "2029-10-17T00:00:00.000Z" usersThreshold: 100 clientsThreshold: 100 + # Allow group-wise migration by clients + allowManualMigration: false lockStatus: locked limitedEventFanout: defaults: @@ -341,10 +345,6 @@ galley: defaults: status: disabled lockStatus: locked - backgroundEffects: - defaults: - status: disabled - lockStatus: locked aws: region: "eu-west-1" proxy: {} @@ -518,6 +518,14 @@ cannon: repository: quay.io/wire/nginz tag: do-not-use pullPolicy: IfNotPresent + # Image for the cannon-configurator initContainer, which only runs a + # single `echo` into a shared volume. Override repository to pull from a + # mirror registry, e.g. my-mirror.example/library/alpine + # renovate: datasource=docker depName=alpine + configuratorImage: + repository: alpine + tag: "3.24.1" + pullPolicy: IfNotPresent config: logLevel: Info logFormat: StructuredJSON @@ -1001,6 +1009,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 users + # It's important to set `settings.postgresMigration.users` to `migration-to-postgresql` + # before starting the migration. + migrateUsers: 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 fa89f36beca..9154c97ea73 100644 --- a/docs/src/developer/reference/config-options.md +++ b/docs/src/developer/reference/config-options.md @@ -249,6 +249,28 @@ The lock status for individual teams can be changed via the internal API (`PUT / The feature status for individual teams can be changed via the public API (if the feature is unlocked). +### Meetings validity and past-edit periods + +`settings.meetings.validityPeriod` (default `48h`) is how long a meeting stays +alive (readable and editable) after its effective end time — its `end_time`, or +the end of its recurrence window for recurring meetings; open-ended recurring +meetings never expire. `settings.meetings.pastEditPeriod` (default `24h`) bounds how +far into the past `PUT /meetings/{domain}/{id}` may move a meeting's +`start_time`/`end_time`, so past and ongoing meetings can be corrected to what +actually happened. Only provided time values are checked against this cutoff; +unchanged stored times are not re-validated — but the effective times (provided +or stored) must still satisfy `end_time > start_time`. Galley refuses to start +if `pastEditPeriod` is negative or greater than `validityPeriod`, so a meeting +edited to past times stays inside the validity window and remains visible and editable. + +```yaml +# galley.yaml +settings: + meetings: + validityPeriod: "48h" + pastEditPeriod: "24h" +``` + ### Meetings email sender and transport The optional `settings.meetings.email` block enables emailing meeting @@ -282,6 +304,13 @@ points at the path where the SMTP password is read, and path into `transport.smtpCredentials.smtpPassword`, the same pattern Brig uses for `smtp.passwordFile`. +### Meetings time settings + +The `galley.config.settings.meetings.legacyTimeZone` Helm value is an IANA time +zone id (e.g. `"Europe/Berlin"`, the default) used as the `tzid` for meetings +created by legacy clients (< V17), which send an `end_time` instead of the V17 +`duration` + `tzid` fields. It has no effect on V17+ clients. + ### Meetings Premium (deprecated) > **Deprecated (WPB-26771).** The `meetingsPremium` feature flag no longer @@ -300,19 +329,20 @@ The aggregate list endpoints (`GET /feature-configs`, `GET /teams/:tid/features`) continue to include `meetingsPremium` at all API versions, including v17. -### Background Effects - -The `backgroundEffects` feature flag controls whether background effects are available in meetings. It is disabled and locked by default. If you want a different configuration, use the following syntax: -```yaml -backgroundEffects: - defaults: - status: disabled|enabled - lockStatus: locked|unlocked -``` +### Background Effects (deprecated) -The lock status for individual teams can be changed via the internal API (`PUT /i/teams/:tid/features/backgroundEffects/(un)?locked`). +> **Deprecated (WPB-27912).** The `backgroundEffects` feature flag no longer +> affects meeting behaviour. The flag, its data type and its public/internal +> endpoints are retained for backward compatibility and are scheduled for +> removal in a future release. -The feature status for individual teams can be changed via the public API (if the feature is unlocked). +The flag now defaults to **enabled and locked** and the Helm configuration +override has been removed (operators can no longer change it via Helm). The +`GET/PUT /teams/:tid/features/backgroundEffects` and internal lock-status +endpoints return 404 at API version v17; they remain available through v16. +The aggregate endpoints `GET /feature-configs` and +`GET /teams/:tid/features` continue to include `backgroundEffects` at all API +versions, including v17. ### File Sharing @@ -362,6 +392,21 @@ The settings mean: - `deletionTimeoutDuration`: how long to keep an adminless conversation before it is deleted. - `reminderTimeoutDurations`: when before deletion reminder notifications should be sent. +In federated conversations, automatic senderless deletion is skipped when the +conversation contains remote members because the corresponding system delete +event cannot yet be sent safely to the remote backend. This applies both when +the feature is enabled and existing conversations are scanned without an +origin user, and when a previously scheduled senderless deletion job runs. +Reminders for a skipped deletion are also skipped because they would be +misleading. The skipped deletion is logged at info level. + +Autopromotion still runs because the conversation-owning backend stores the +authoritative member roles. Remote clients may miss the immediate senderless +member-update notification, but a subsequent conversation fetch obtains the +current role from the owning backend. Member updates and deletions with an +origin user continue to use the existing ordinary federation events and are +not skipped. + Durations are strings with a number and a unit suffix. Supported units are `us`, `ms`, `s`, `m`, `h`, `d`, and `w`. It is **not** recommended or supported to set these below a day in production environments. Feature responses, including `GET /feature-configs`, `GET /teams/:tid/features`, and `GET /teams/:tid/features/preventAdminlessGroups`, include the duration fields: @@ -380,7 +425,7 @@ Feature responses, including `GET /feature-configs`, `GET /teams/:tid/features`, From a client's perspective, API versioning works like this: -- API version V17 and newer should send the duration fields to `PUT /teams/:tid/features/preventAdminlessGroups`. +- API version V18 and newer should send the duration fields to `PUT /teams/:tid/features/preventAdminlessGroups`. - Feature responses include the duration fields for clients to read. The lock status for individual teams can be changed via the internal API (`PUT /i/teams/:tid/features/preventAdminlessGroups/(un)?locked`). @@ -2083,6 +2128,7 @@ The current settings and their background-worker flags are: - `conversationCodes` -> `migrateConversationCodes` - `teamFeatures` -> `migrateTeamFeatures` - `domainRegistration` -> `migrateDomainRegistration` +- `user` -> `migrateUsers` **Migration pattern per migration setting** @@ -2101,13 +2147,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: migration-to-postgresql + user: migration-to-postgresql background-worker: config: migrateConversations: false migrateConversationCodes: false migrateTeamFeatures: false - migrateDomainRegistration: false + migrateDomainRegistration: false + migrateUsers: false ``` This change should restart the affected pods, and new writes will follow the @@ -2121,7 +2169,8 @@ The current settings and their background-worker flags are: migrateConversations: true migrateConversationCodes: true migrateTeamFeatures: true - migrateDomainRegistration: true + migrateDomainRegistration: true + migrateUsers: true ``` During migration, Cassandra rows are not deleted. Writes and migration share @@ -2137,6 +2186,16 @@ 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` + - `user`: `wire_user_migration_finished` + + > ⚠️ For user migrations please watch the logs for `Invalid user found, + > skipping`. This would be accompanied by an error which is either + > `UserHasNoName` or `UserHasNoActivated`. These users are invalid and all + > interactions with them were resulting in errors. If these warnings are + > ignored, these users will stop existing in the system. If these users are + > to be saved, the operator must insert some value as `name` and/or + > `activated` and then re-trigger the migration **after** the background + > worker finishes migrating the valid users. 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 @@ -2149,13 +2208,15 @@ The current settings and their background-worker flags are: conversation: postgresql conversationCodes: postgresql teamFeatures: postgresql - domainRegistration: cassandra + domainRegistration: postgresql + user: postgresql background-worker: config: migrateConversations: false migrateConversationCodes: false migrateTeamFeatures: false - migrateDomainRegistration: false + migrateDomainRegistration: false + migrateUsers: false ``` **How to run migrations independently or in batches** diff --git a/postgres-schema.sql b/postgres-schema.sql index c96cc12aa37..2d3df5fb27f 100644 --- a/postgres-schema.sql +++ b/postgres-schema.sql @@ -1481,6 +1481,7 @@ CREATE TABLE public.meetings ( ALTER TABLE public.meetings OWNER TO "wire-server"; +-- -- Name: mls_group_member_client; Type: TABLE; Schema: public; Owner: wire-server -- diff --git a/services/background-worker/src/Wire/BackgroundWorker.hs b/services/background-worker/src/Wire/BackgroundWorker.hs index b57ba12df40..6c12b02e816 100644 --- a/services/background-worker/src/Wire/BackgroundWorker.hs +++ b/services/background-worker/src/Wire/BackgroundWorker.hs @@ -78,6 +78,14 @@ run opts galleyOpts = do withNamedLogger "migrate-domain-registration" $ Migrations.domainRegistration opts.migrationOptions else pure $ pure () + cleanupUsersMigration <- + if opts.migrateUsers + then + runAppT env $ + withNamedLogger "migrate-users" $ + Migrations.users opts.migrationOptions + else pure $ pure () + cleanupJobs <- runAppT env $ withNamedLogger "background-job-consumer" $ @@ -89,13 +97,14 @@ run opts galleyOpts = do let cleanup = void $ runConcurrently $ - (,,,,,,,) + (,,,,,,,,) <$> Concurrently cleanupDeadUserNotifWatcher <*> Concurrently cleanupBackendNotifPusher <*> Concurrently cleanupConvMigration <*> Concurrently cleanUpConvCodesMigration <*> Concurrently cleanupTeamFeaturesMigration <*> Concurrently cleanupDomainRegistrationMigration + <*> Concurrently cleanupUsersMigration <*> 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..035460cfc32 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, + migrateUsers :: !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..28c6a789a4a 100644 --- a/services/background-worker/src/Wire/PostgresMigrations.hs +++ b/services/background-worker/src/Wire/PostgresMigrations.hs @@ -24,10 +24,11 @@ import UnliftIO import Wire.BackgroundWorker.Env import Wire.BackgroundWorker.Util import Wire.CodeStore.Migration -import Wire.ConversationStore.Migration +import Wire.ConversationStore.Migration qualified as ConversationStore import Wire.DomainRegistrationStore.Migration import Wire.Migration (MigrationOptions) import Wire.TeamFeatureStore.Migration +import Wire.UserStore.Migration qualified as UserStore conversations :: MigrationOptions -> AppT IO CleanupAction conversations migOpts = do @@ -45,8 +46,8 @@ conversations migOpts = do userMigFailed <- register $ counter $ Prometheus.Info "wire_user_remote_convs_migration_failed" "Whether the migration of remote conversation membership data to Postgresql has failed" userMigDuration <- register $ vector "outcome" $ histogram (Prometheus.Info "wire_user_remote_convs_migration_duration_seconds" "Duration of remote conversation membership migration attempts") defaultBuckets - convLoop <- async . lift $ migrateConvsLoop migOpts cassClient pgPool logger convMigCounter convMigFinished convMigFailed convMigDuration - userLoop <- async . lift $ migrateUsersLoop migOpts cassClient pgPool logger userMigCounter userMigFinished userMigFailed userMigDuration + convLoop <- async . lift $ ConversationStore.migrateConvsLoop migOpts cassClient pgPool logger convMigCounter convMigFinished convMigFailed convMigDuration + userLoop <- async . lift $ ConversationStore.migrateUsersLoop migOpts cassClient pgPool logger userMigCounter userMigFinished userMigFailed userMigDuration Log.info logger $ Log.msg (Log.val "started conversation migration") pure $ do @@ -107,3 +108,21 @@ domainRegistration migOpts = do pure $ do Log.info logger $ Log.msg (Log.val "cancelling domain registration migration") cancel migrationLoop + +users :: MigrationOptions -> AppT IO CleanupAction +users migOpts = do + cassClient <- asks (.cassandraBrig) + pgPool <- asks (.hasqlPool) + logger <- asks (.logger) + Log.info logger $ Log.msg (Log.val "starting user migration") + count <- register $ counter $ Prometheus.Info "wire_users_migrated_to_pg" "Number of user rows migrated to Postgresql" + finished <- register $ counter $ Prometheus.Info "wire_users_migration_finished" "Whether the user migration to Postgresql is finished successfully" + failed <- register $ counter $ Prometheus.Info "wire_users_migration_failed" "Whether the user migration to Postgresql has failed" + duration <- register $ vector "outcome" $ histogram (Prometheus.Info "wire_users_migration_duration_seconds" "Duration of user migration attempts") defaultBuckets + + migrationLoop <- async . lift $ UserStore.migrateUsersLoop migOpts cassClient pgPool logger count finished failed duration + + Log.info logger $ Log.msg (Log.val "started user migration") + pure $ do + Log.info logger $ Log.msg (Log.val "cancelling user migration") + cancel migrationLoop From c2ba0620001877bef11f17ba2dbfa1dc22568108 Mon Sep 17 00:00:00 2001 From: Gautier DI FOLCO Date: Thu, 3 Sep 2026 12:22:26 +0200 Subject: [PATCH 6/9] Hello CI From e2d1199e308534f63c2b47af90a6e2781c00fb9e Mon Sep 17 00:00:00 2001 From: Gautier DI FOLCO Date: Thu, 3 Sep 2026 14:34:05 +0200 Subject: [PATCH 7/9] WPB-22954: connect users in commit-lock test randomUser users are not connected, so galley rejects the commit bundle with 403 not-connected. Use createAndConnectUsers, matching the old testCommitLock and the other integration-package MLS tests. --- integration/test/Test/MLS/CommitLock.hs | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/integration/test/Test/MLS/CommitLock.hs b/integration/test/Test/MLS/CommitLock.hs index 3b34ff14f9b..f6ef6938d3d 100644 --- a/integration/test/Test/MLS/CommitLock.hs +++ b/integration/test/Test/MLS/CommitLock.hs @@ -26,15 +26,13 @@ import Testlib.Prelude -- interpreter. A leaked lock would fail the second commit. testMLSCommitLock :: (HasCallStack) => App () testMLSCommitLock = do - alice <- randomUser OwnDomain def + [alice, bob, charlie] <- createAndConnectUsers [OwnDomain, OwnDomain, OwnDomain] alice1 <- createMLSClient def alice - bob <- randomUser OwnDomain def bob1 <- createMLSClient def bob void $ uploadNewKeyPackage def bob1 convId <- createNewGroup def alice1 void $ createAddCommit alice1 convId [bob] >>= sendAndConsumeCommitBundle - charlie <- randomUser OwnDomain def charlie1 <- createMLSClient def charlie void $ uploadNewKeyPackage def charlie1 void $ createAddCommit alice1 convId [charlie] >>= sendAndConsumeCommitBundle From 9cd07ec4d6d25a6d962d05a9bb9eb610853c8583 Mon Sep 17 00:00:00 2001 From: Gautier DI FOLCO Date: Thu, 3 Sep 2026 15:45:12 +0200 Subject: [PATCH 8/9] Hello CI From 4bf7ce96fc0d4153a4daee1086a22d8691a2620f Mon Sep 17 00:00:00 2001 From: Gautier DI FOLCO Date: Thu, 3 Sep 2026 16:43:11 +0200 Subject: [PATCH 9/9] Hello CI