[WPB-28375] Move sync-user-group background jobs from RabbitMQ to Arbiter

changelog.d/2-features/WPB-28375-user-groups-arbiter

@@ -0,0 +1,9 @@
+Move the sync-user-group family of background jobs from the RabbitMQ `background-jobs` queue to the Arbiter PostgreSQL job queue (new `user-groups` queue/table). brig now enqueues sync jobs via the shared Postgres pool with per-user-group serialization and 3 attempts; background-worker consumes them in a third Arbiter worker pool. The RabbitMQ jobs publisher is removed from brig; the RabbitMQ jobs consumer is KEPT in this release so that jobs still sitting in the `background-jobs` queue when you upgrade are drained instead of dropped. backend-notification push and dead-user-notification watching remain on RabbitMQ.
+
+Rollout: THIS RELEASE MUST NOT BE SKIPPED.
+
+1. `helm upgrade` to this release as usual. From the moment brig starts, new user-group sync jobs go to Arbiter; background-worker also keeps consuming leftover jobs from the RabbitMQ `background-jobs` queue.
+2. Wait until the `background-jobs` queue is empty before upgrading further. brig no longer publishes to it, and the consumer only re-publishes child jobs of leftover parents to the same queue, so the count never grows and trends to zero. Check with either:
+   - `kubectl exec <rabbitmq-pod> -- rabbitmqctl list_queues name messages | grep background-jobs`, or
+   - the RabbitMQ management HTTP API: `GET /api/queues/<vhost>/background-jobs` and read the `messages` field.
+3. Upgrade to the next release, which removes the RabbitMQ consumer. Skipping this release (jumping from the previous one to the next) would drop any jobs still in the queue; those sync jobs are idempotent reconciliations re-triggered by the next SCIM write, but the drain window avoids relying on that.

docs/src/developer/reference/config-options.md

@@ -2322,15 +2322,13 @@ The job runner uses polling rather than LISTEN/NOTIFY. It therefore does not
 open a separate listener connection; new jobs are discovered according to
 `jobs.pollInterval`.
 
-`backgroundJobs` and `jobs` configure different job systems. The
-`backgroundJobs` consumer receives immediate user-group synchronization jobs
-from RabbitMQ and controls their in-process concurrency, timeout, and retry
-behavior. `jobs` runs Arbiter-backed PostgreSQL jobs that may be
-scheduled for a future time, including recurring jobs, and controls their
-dispatcher, worker-pool, visibility, retry, and reaper behavior. The systems
-are separate because they currently use different transports and execution
-semantics. They could be merged in the future if the user-group jobs are
-migrated to Arbiter.
+`jobs` configures the Arbiter-backed PostgreSQL queues (meetings, conversations,
+and user groups), covering immediate and scheduled or recurring jobs, and
+controls their dispatcher, worker-pool, visibility, retry, and reaper behavior.
+`backgroundJobs` configures the legacy RabbitMQ consumer, which this release
+keeps only to drain `background-jobs` queue entries still published by the
+previous release's brig; brig no longer publishes to RabbitMQ, so the queue
+only shrinks, and the consumer will be removed in the next release.
 
 # Required for addressing local vs remote backends
 federationDomain: example.org
@@ -2367,10 +2365,10 @@ Notes
 - RabbitMQ admin fields (`adminHost`, `adminPort`) are templated only when `config.enableFederation` is true.
 - In the Helm charts, `background-worker` reads `postgresMigration` from `galley.config.postgresMigration`.
 - The `migrate...` flags control the corresponding PostgreSQL backfill jobs for the current migration settings; leave them `false` for new installs and after migration.
-- `concurrency`, `jobTimeout`, and `maxAttempts` control parallelism and retry behavior of the consumer.
+- `concurrency`, `jobTimeout`, and `maxAttempts` control parallelism and retry behavior of the legacy RabbitMQ `background-jobs` consumer.
 - `brig` and `gundeck` endpoints default to in-cluster services; override via `background-worker.config.brig` and `.gundeck` if your service DNS/ports differ.
 - `jobs` controls the Arbiter dispatcher, worker, retry, shutdown, and reaper settings. All fields default to the values shown above.
 - `jobs.pollInterval` controls how often the background worker wakes up to check for due jobs.
 - `jobs.workerThreads` controls the number of worker threads in each job queue. The default is `1`; increasing it allows jobs in that queue to run in parallel when their group keys permit it.
-- Both job queues share the same PostgreSQL pool. Increasing `jobs.workerThreads` can increase the number of connections needed when more jobs run concurrently, but it does not create a permanently dedicated connection per thread or queue.
+- All three job queues share the same PostgreSQL pool. Increasing `jobs.workerThreads` can increase the number of connections needed when more jobs run concurrently, but it does not create a permanently dedicated connection per thread or queue.
 - The job runner is poll-only and does not require an additional PostgreSQL listener connection.

libs/wire-api/src/Wire/API/BackgroundJobs.hs

@@ -73,7 +73,7 @@ data SyncUserGroupAndChannel = SyncUserGroupAndChannel
     actor :: Maybe UserId
   }
   deriving (Show, Eq, Generic)
-  deriving (Aeson.ToJSON, Aeson.FromJSON) via (Schema SyncUserGroupAndChannel)
+  deriving (Aeson.ToJSON, Aeson.FromJSON, S.ToSchema) via (Schema SyncUserGroupAndChannel)
   deriving (Arbitrary) via GenericUniform SyncUserGroupAndChannel
 
 instance ToSchema SyncUserGroupAndChannel where
@@ -91,7 +91,7 @@ data SyncUserGroup = SyncUserGroup
     actor :: Maybe UserId
   }
   deriving (Show, Eq, Generic)
-  deriving (Aeson.ToJSON, Aeson.FromJSON) via (Schema SyncUserGroup)
+  deriving (Aeson.ToJSON, Aeson.FromJSON, S.ToSchema) via (Schema SyncUserGroup)
   deriving (Arbitrary) via GenericUniform SyncUserGroup
 
 instance ToSchema SyncUserGroup where

libs/wire-api/src/Wire/API/Jobs.hs

@@ -36,6 +36,7 @@ import Data.Text as Text
 import GHC.TypeLits
 import Imports
 import Test.QuickCheck (oneof)
+import Wire.API.BackgroundJobs (SyncUserGroup, SyncUserGroupAndChannel)
 import Wire.Arbitrary (Arbitrary (..), GenericUniform (..))
 
 -- | The queue/table for jobs that operate on meetings.
@@ -50,6 +51,12 @@ type ConversationsQueueName = "conversations"
 conversationsQueueName :: Text
 conversationsQueueName = Text.pack $ symbolVal (Proxy @ConversationsQueueName)
 
+-- | The queue/table for jobs that operate on user groups.
+type UserGroupsQueueName = "user-groups"
+
+userGroupsQueueName :: Text
+userGroupsQueueName = Text.pack $ symbolVal (Proxy @UserGroupsQueueName)
+
 -- | Empty payload because the schedule itself carries all execution context.
 data MeetingsCleanupJob = MeetingsCleanupJob
   deriving stock (Eq, Generic, Show)
@@ -247,8 +254,97 @@ deriving via (Schema ConversationsJobPayload) instance S.ToSchema ConversationsJ
 instance Arbitrary ConversationsJobPayload where
   arbitrary = oneof [AdminlessDeletion <$> arbitrary, AdminlessReminder <$> arbitrary]
 
+-- | Payload for synchronising a user group (without its channel contents).
+data UserGroupsSyncUserGroupJob = UserGroupsSyncUserGroupJob
+  { userGroupsSyncUserGroupJobRequestId :: RequestId,
+    userGroupsSyncUserGroupJobData :: SyncUserGroup
+  }
+  deriving stock (Eq, Generic, Show)
+  deriving (ToJSON, FromJSON, S.ToSchema) via (Schema UserGroupsSyncUserGroupJob)
+
+instance Arbitrary UserGroupsSyncUserGroupJob where
+  arbitrary = UserGroupsSyncUserGroupJob <$> arbitrary <*> arbitrary
+
+instance ToSchema UserGroupsSyncUserGroupJob where
+  schema =
+    object $
+      UserGroupsSyncUserGroupJob
+        <$> (.userGroupsSyncUserGroupJobRequestId) .= field "request_id" schema
+        <*> (.userGroupsSyncUserGroupJobData) .= field "data" schema
+
+-- | Payload for synchronising a user group together with one of its channels.
+data UserGroupsSyncUserGroupAndChannelJob = UserGroupsSyncUserGroupAndChannelJob
+  { userGroupsSyncUserGroupAndChannelJobRequestId :: RequestId,
+    userGroupsSyncUserGroupAndChannelJobData :: SyncUserGroupAndChannel
+  }
+  deriving stock (Eq, Generic, Show)
+  deriving (ToJSON, FromJSON, S.ToSchema) via (Schema UserGroupsSyncUserGroupAndChannelJob)
+
+instance Arbitrary UserGroupsSyncUserGroupAndChannelJob where
+  arbitrary = UserGroupsSyncUserGroupAndChannelJob <$> arbitrary <*> arbitrary
+
+instance ToSchema UserGroupsSyncUserGroupAndChannelJob where
+  schema =
+    object $
+      UserGroupsSyncUserGroupAndChannelJob
+        <$> (.userGroupsSyncUserGroupAndChannelJobRequestId) .= field "request_id" schema
+        <*> (.userGroupsSyncUserGroupAndChannelJobData) .= field "data" schema
+
+-- | Payload persisted in the user-groups queue. Keep the type tags and nested
+-- data shapes stable when changing job payloads.
+data UserGroupsJobPayload
+  = UserGroupsSyncUserGroup UserGroupsSyncUserGroupJob
+  | UserGroupsSyncUserGroupAndChannel UserGroupsSyncUserGroupAndChannelJob
+  deriving stock (Eq, Generic, Show)
+
+data UserGroupsJobPayloadTag
+  = UserGroupsSyncUserGroupTag
+  | UserGroupsSyncUserGroupAndChannelTag
+  deriving stock (Eq, Ord, Bounded, Enum, Show, Generic)
+  deriving (Arbitrary) via GenericUniform UserGroupsJobPayloadTag
+
+instance ToSchema UserGroupsJobPayloadTag where
+  schema =
+    enum @Text $
+      mconcat
+        [ element "sync_user_group" UserGroupsSyncUserGroupTag,
+          element "sync_user_group_and_channel" UserGroupsSyncUserGroupAndChannelTag
+        ]
+
+makePrisms ''UserGroupsJobPayload
+
+userGroupsJobPayloadObjectSchema :: ObjectSchema SwaggerDoc UserGroupsJobPayload
+userGroupsJobPayloadObjectSchema = taggedJobPayloadObjectSchema toTag toSchema
+  where
+    toTag :: UserGroupsJobPayload -> UserGroupsJobPayloadTag
+    toTag = \case
+      UserGroupsSyncUserGroup {} -> UserGroupsSyncUserGroupTag
+      UserGroupsSyncUserGroupAndChannel {} -> UserGroupsSyncUserGroupAndChannelTag
+
+    toSchema :: UserGroupsJobPayloadTag -> ObjectSchema SwaggerDoc UserGroupsJobPayload
+    toSchema = \case
+      UserGroupsSyncUserGroupTag -> tag _UserGroupsSyncUserGroup (field "data" schema)
+      UserGroupsSyncUserGroupAndChannelTag -> tag _UserGroupsSyncUserGroupAndChannel (field "data" schema)
+
+instance ToSchema UserGroupsJobPayload where
+  schema = object userGroupsJobPayloadObjectSchema
+
+deriving via (Schema UserGroupsJobPayload) instance FromJSON UserGroupsJobPayload
+
+deriving via (Schema UserGroupsJobPayload) instance ToJSON UserGroupsJobPayload
+
+deriving via (Schema UserGroupsJobPayload) instance S.ToSchema UserGroupsJobPayload
+
+instance Arbitrary UserGroupsJobPayload where
+  arbitrary =
+    oneof
+      [ UserGroupsSyncUserGroup <$> arbitrary,
+        UserGroupsSyncUserGroupAndChannel <$> arbitrary
+      ]
+
 -- | Registry for the jobs we expose via Arbiter.
 type JobRegistry =
   '[ Queue MeetingsQueueName MeetingsJobPayload,
-     Queue ConversationsQueueName ConversationsJobPayload
+     Queue ConversationsQueueName ConversationsJobPayload,
+     Queue UserGroupsQueueName UserGroupsJobPayload
    ]

libs/wire-subsystems/src/Wire/BackgroundJobsPublisher.hs

@@ -19,11 +19,10 @@
 
 module Wire.BackgroundJobsPublisher where
 
-import Data.Id
 import Polysemy
 import Wire.API.BackgroundJobs (BackgroundJobPayload)
 
 data BackgroundJobPublisher m a where
-  PublishJob :: JobId -> BackgroundJobPayload -> BackgroundJobPublisher m ()
+  PublishJob :: BackgroundJobPayload -> BackgroundJobPublisher m ()
 
 makeSem ''BackgroundJobPublisher

libs/wire-subsystems/src/Wire/BackgroundJobsPublisher/Arbiter.hs

@@ -0,0 +1,82 @@
+{-# LANGUAGE OverloadedRecordDot #-}
+{-# LANGUAGE RecordWildCards #-}
+{-# LANGUAGE TypeApplications #-}
+
+-- This file is part of the Wire Server implementation.
+--
+-- Copyright (C) 2026 Wire Swiss GmbH <opensource@wire.com>
+--
+-- This program is free software: you can redistribute it and/or modify it under
+-- the terms of the GNU Affero General Public License as published by the Free
+-- Software Foundation, either version 3 of the License, or (at your option) any
+-- later version.
+--
+-- This program is distributed in the hope that it will be useful, but WITHOUT
+-- ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+-- FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License
+-- for more details.
+--
+-- You should have received a copy of the GNU Affero General Public License along
+-- with this program. If not, see <https://www.gnu.org/licenses/>.
+
+module Wire.BackgroundJobsPublisher.Arbiter
+  ( interpretBackgroundJobPublisherArbiter,
+  )
+where
+
+import Arbiter.Core qualified as ArbiterCore
+import Data.Id
+import Hasql.Pool.Extended qualified as HasqlPoolExt
+import Imports
+import Polysemy
+import Polysemy.Input (input)
+import Wire.API.BackgroundJobs
+import Wire.API.Jobs
+import Wire.BackgroundJobsPublisher (BackgroundJobPublisher (..))
+import Wire.JobSubsystem (JobSubsystemConfig (..))
+import Wire.JobSubsystem.ArbiterAdapter
+import Wire.Postgres (PGConstraints)
+
+interpretBackgroundJobPublisherArbiter ::
+  (PGConstraints r) =>
+  RequestId ->
+  JobSubsystemConfig ->
+  InterpreterFor BackgroundJobPublisher r
+interpretBackgroundJobPublisherArbiter requestId conf =
+  interpret
+    \case
+      PublishJob payload -> publishJob requestId conf payload
+
+publishJob ::
+  (PGConstraints r) =>
+  RequestId ->
+  JobSubsystemConfig ->
+  BackgroundJobPayload ->
+  Sem r ()
+publishJob requestId JobSubsystemConfig {..} = \case
+  BackgroundJobSyncUserGroup syncUserGroup ->
+    insertUserGroupsJob
+      jobSubsystemSchemaName
+      (UserGroupsSyncUserGroup (UserGroupsSyncUserGroupJob requestId syncUserGroup))
+      syncUserGroup.userGroupId
+  BackgroundJobSyncUserGroupAndChannel syncUserGroupAndChannel ->
+    insertUserGroupsJob
+      jobSubsystemSchemaName
+      (UserGroupsSyncUserGroupAndChannel (UserGroupsSyncUserGroupAndChannelJob requestId syncUserGroupAndChannel))
+      syncUserGroupAndChannel.userGroupId
+
+insertUserGroupsJob ::
+  (PGConstraints r) =>
+  Text ->
+  UserGroupsJobPayload ->
+  UserGroupId ->
+  Sem r ()
+insertUserGroupsJob schemaName payload userGroupId = do
+  pool <- input @HasqlPoolExt.Pool
+  let arbiterEnv = mkNewWireArbiterEnv schemaName pool
+      groupKey = "user-group-sync:" <> idToText userGroupId
+      arbiterJob =
+        (ArbiterCore.defaultGroupedJob groupKey payload)
+          { ArbiterCore.maxAttempts = Just 3
+          }
+  embed $ void $ runWireArbiter arbiterEnv $ ArbiterCore.insertJob @UserGroupsJobPayload @(WireArbiter JobRegistry) arbiterJob

libs/wire-subsystems/src/Wire/BackgroundJobsPublisher/RabbitMQ.hs

@@ -18,7 +18,7 @@
 module Wire.BackgroundJobsPublisher.RabbitMQ where
 
 import Data.Aeson qualified as Aeson
-import Data.Id (JobId, RequestId (..), idToText)
+import Data.Id (JobId, RequestId (..), idToText, randomId)
 import Data.Text.Encoding qualified as T
 import Imports
 import Network.AMQP qualified as Q
@@ -33,7 +33,8 @@ interpretBackgroundJobPublisherRabbitMQ ::
   InterpreterFor BackgroundJobPublisher r
 interpretBackgroundJobPublisherRabbitMQ requestId channelMVar =
   interpret $ \case
-    PublishJob jobId jobPayload -> do
+    PublishJob jobPayload -> do
+      jobId <- randomId
       channel <- readMVar channelMVar
       publishJob requestId channel jobId jobPayload
 

libs/wire-subsystems/src/Wire/BackgroundJobsRunner.hs

@@ -20,9 +20,9 @@
 module Wire.BackgroundJobsRunner where
 
 import Polysemy
-import Wire.API.BackgroundJobs (BackgroundJob)
+import Wire.API.BackgroundJobs (BackgroundJobPayload)
 
 data BackgroundJobRunner m a where
-  RunJob :: BackgroundJob -> BackgroundJobRunner m ()
+  RunJob :: BackgroundJobPayload -> BackgroundJobRunner m ()
 
 makeSem ''BackgroundJobRunner

libs/wire-subsystems/src/Wire/BackgroundJobsRunner/Interpreter.hs

@@ -45,7 +45,6 @@ import Wire.BackgroundJobsPublisher
 import Wire.BackgroundJobsRunner (BackgroundJobRunner (..))
 import Wire.ConversationStore (ConversationStore, upsertMembers)
 import Wire.ConversationSubsystem
-import Wire.Sem.Random
 import Wire.StoredConversation
 import Wire.UserGroupStore (UserGroupStore, getUserGroup, getUserGroupChannels)
 import Wire.UserList (toUserList)
@@ -56,7 +55,6 @@ interpretBackgroundJobRunner ::
     Member (Input (Local ())) r,
     Member ConversationStore r,
     Member ConversationSubsystem r,
-    Member Random r,
     Member TinyLog r
   ) =>
   InterpreterFor BackgroundJobRunner r
@@ -69,12 +67,11 @@ runBackgroundJob ::
     Member (Input (Local ())) r,
     Member ConversationStore r,
     Member ConversationSubsystem r,
-    Member Random r,
     Member TinyLog r
   ) =>
-  BackgroundJob ->
+  BackgroundJobPayload ->
   Sem r ()
-runBackgroundJob job = case job.payload of
+runBackgroundJob = \case
   BackgroundJobSyncUserGroupAndChannel payload -> runSyncUserGroupAndChannel payload
   BackgroundJobSyncUserGroup payload -> runSyncUserGroup payload
 
@@ -146,7 +143,6 @@ runSyncUserGroupAndChannel (SyncUserGroupAndChannel {..}) = do
 runSyncUserGroup ::
   ( Member UserGroupStore r,
     Member BackgroundJobPublisher r,
-    Member Random r,
     Member TinyLog r
   ) =>
   SyncUserGroup ->
@@ -161,5 +157,4 @@ runSyncUserGroup SyncUserGroup {..} = do
   let channels = fromMaybe mempty mChannels
   for_ channels $ \convId -> do
     let syncUserGroupAndChannel = SyncUserGroupAndChannel {..}
-    jobId <- newId
-    publishJob jobId (BackgroundJobSyncUserGroupAndChannel syncUserGroupAndChannel)
+    publishJob (BackgroundJobSyncUserGroupAndChannel syncUserGroupAndChannel)