diff --git a/changelog.d/2-features/WPB-28375-user-groups-arbiter b/changelog.d/2-features/WPB-28375-user-groups-arbiter new file mode 100644 index 00000000000..c3d3e2ee5a3 --- /dev/null +++ b/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 -- rabbitmqctl list_queues name messages | grep background-jobs`, or + - the RabbitMQ management HTTP API: `GET /api/queues//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. diff --git a/docs/src/developer/reference/config-options.md b/docs/src/developer/reference/config-options.md index c1b48558c7c..acb9e6b778a 100644 --- a/docs/src/developer/reference/config-options.md +++ b/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. diff --git a/libs/wire-api/src/Wire/API/BackgroundJobs.hs b/libs/wire-api/src/Wire/API/BackgroundJobs.hs index b96f2cd7505..ac99d5dd33e 100644 --- a/libs/wire-api/src/Wire/API/BackgroundJobs.hs +++ b/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 diff --git a/libs/wire-api/src/Wire/API/Jobs.hs b/libs/wire-api/src/Wire/API/Jobs.hs index f60d60ca74f..14468710ac1 100644 --- a/libs/wire-api/src/Wire/API/Jobs.hs +++ b/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 ] diff --git a/libs/wire-subsystems/src/Wire/BackgroundJobsPublisher.hs b/libs/wire-subsystems/src/Wire/BackgroundJobsPublisher.hs index 3c3012e876c..43d9fc9028e 100644 --- a/libs/wire-subsystems/src/Wire/BackgroundJobsPublisher.hs +++ b/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 diff --git a/libs/wire-subsystems/src/Wire/BackgroundJobsPublisher/Arbiter.hs b/libs/wire-subsystems/src/Wire/BackgroundJobsPublisher/Arbiter.hs new file mode 100644 index 00000000000..cb94cfb362a --- /dev/null +++ b/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 +-- +-- 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.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 diff --git a/libs/wire-subsystems/src/Wire/BackgroundJobsPublisher/RabbitMQ.hs b/libs/wire-subsystems/src/Wire/BackgroundJobsPublisher/RabbitMQ.hs index 1b3e9cb273b..6a6719b4fff 100644 --- a/libs/wire-subsystems/src/Wire/BackgroundJobsPublisher/RabbitMQ.hs +++ b/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 diff --git a/libs/wire-subsystems/src/Wire/BackgroundJobsRunner.hs b/libs/wire-subsystems/src/Wire/BackgroundJobsRunner.hs index d54e2d64a07..6bc2a71f2d7 100644 --- a/libs/wire-subsystems/src/Wire/BackgroundJobsRunner.hs +++ b/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 diff --git a/libs/wire-subsystems/src/Wire/BackgroundJobsRunner/Interpreter.hs b/libs/wire-subsystems/src/Wire/BackgroundJobsRunner/Interpreter.hs index 8fc5abdc680..466f489426f 100644 --- a/libs/wire-subsystems/src/Wire/BackgroundJobsRunner/Interpreter.hs +++ b/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) diff --git a/libs/wire-subsystems/src/Wire/UserGroupSubsystem/Interpreter.hs b/libs/wire-subsystems/src/Wire/UserGroupSubsystem/Interpreter.hs index 59565c35747..e08122b5da8 100644 --- a/libs/wire-subsystems/src/Wire/UserGroupSubsystem/Interpreter.hs +++ b/libs/wire-subsystems/src/Wire/UserGroupSubsystem/Interpreter.hs @@ -49,7 +49,6 @@ import Wire.BackgroundJobsPublisher import Wire.Error import Wire.GalleyAPIAccess (GalleyAPIAccess, internalGetConversation) import Wire.NotificationSubsystem -import Wire.Sem.Random qualified as Random import Wire.TeamSubsystem import Wire.UserGroupStore qualified as Store import Wire.UserGroupSubsystem (UserGroupSubsystem (..)) @@ -57,8 +56,7 @@ import Wire.UserSubsystem (UserSubsystem, getLocalUserProfiles, getUserTeam) import Wire.Util interpretUserGroupSubsystem :: - ( Member Random.Random r, - Member UserSubsystem r, + ( Member UserSubsystem r, Member (Error UserGroupSubsystemError) r, Member Store.UserGroupStore r, Member (Input (Local ())) r, @@ -106,8 +104,7 @@ userGroupSubsystemErrorToHttpError = UserGroupManagedByMismatch -> errorToWai @E.UserGroupManagedByMismatch createUserGroup :: - ( Member Random.Random r, - Member UserSubsystem r, + ( Member UserSubsystem r, Member (Error UserGroupSubsystemError) r, Member Store.UserGroupStore r, Member (Input (Local ())) r, @@ -130,7 +127,6 @@ createUserGroupFullImpl :: Member (Input (Local ())) r, Member NotificationSubsystem r, Member TeamSubsystem r, - Member Random.Random r, Member BackgroundJobPublisher r ) => ManagedBy -> @@ -362,8 +358,7 @@ deleteGroupManagedImpl managedBy team groupId = do throw UserGroupManagedByMismatch addUser :: - ( Member Random.Random r, - Member UserSubsystem r, + ( Member UserSubsystem r, Member Store.UserGroupStore r, Member (Error UserGroupSubsystemError) r, Member NotificationSubsystem r, @@ -387,8 +382,7 @@ addUser adder groupId addeeId = do triggerSyncUserGroup team (Just adder) groupId addUsers :: - ( Member Random.Random r, - Member UserSubsystem r, + ( Member UserSubsystem r, Member Store.UserGroupStore r, Member (Error UserGroupSubsystemError) r, Member NotificationSubsystem r, @@ -416,8 +410,7 @@ addUsers adder groupId addeeIds = do triggerSyncUserGroup team (Just adder) groupId updateUsers :: - ( Member Random.Random r, - Member UserSubsystem r, + ( Member UserSubsystem r, Member Store.UserGroupStore r, Member (Error UserGroupSubsystemError) r, Member NotificationSubsystem r, @@ -437,7 +430,6 @@ updateUsersNoAccessControl :: Member (Error UserGroupSubsystemError) r, Member NotificationSubsystem r, Member TeamSubsystem r, - Member Random.Random r, Member BackgroundJobPublisher r ) => TeamId -> @@ -457,8 +449,7 @@ updateUsersNoAccessControl teamId mbUpdater groupId uids = do triggerSyncUserGroup teamId mbUpdater groupId removeUser :: - ( Member Random.Random r, - Member UserSubsystem r, + ( Member UserSubsystem r, Member Store.UserGroupStore r, Member (Error UserGroupSubsystemError) r, Member NotificationSubsystem r, @@ -518,8 +509,7 @@ removeUserFromAllGroups uid tid = do } updateChannels :: - ( Member Random.Random r, - Member UserSubsystem r, + ( Member UserSubsystem r, Member Store.UserGroupStore r, Member (Error UserGroupSubsystemError) r, Member TeamSubsystem r, @@ -553,23 +543,19 @@ updateChannels appendOnly performer groupId channelIds = do ] triggerSyncUserGroup :: - ( Member Random.Random r, - Member BackgroundJobPublisher r - ) => + (Member BackgroundJobPublisher r) => TeamId -> Maybe UserId -> UserGroupId -> Sem r () -triggerSyncUserGroup teamId actor userGroupId = do - jobId <- Random.newId - publishJob jobId $ BackgroundJobSyncUserGroup SyncUserGroup {..} +triggerSyncUserGroup teamId actor userGroupId = + publishJob $ BackgroundJobSyncUserGroup SyncUserGroup {..} resetUserGroupInternal :: ( Member Store.UserGroupStore r, Member (Error UserGroupSubsystemError) r, Member TeamSubsystem r, Member NotificationSubsystem r, - Member Random.Random r, Member BackgroundJobPublisher r ) => UpdateGroupInternalRequest -> diff --git a/libs/wire-subsystems/wire-subsystems.cabal b/libs/wire-subsystems/wire-subsystems.cabal index 220d4252fb5..e32d082bc34 100644 --- a/libs/wire-subsystems/wire-subsystems.cabal +++ b/libs/wire-subsystems/wire-subsystems.cabal @@ -236,6 +236,7 @@ library Wire.BackendNotificationQueueAccess Wire.BackendNotificationQueueAccess.RabbitMq Wire.BackgroundJobsPublisher + Wire.BackgroundJobsPublisher.Arbiter Wire.BackgroundJobsPublisher.RabbitMQ Wire.BackgroundJobsRunner Wire.BackgroundJobsRunner.Interpreter diff --git a/services/background-worker/src/Wire/BackgroundWorker.hs b/services/background-worker/src/Wire/BackgroundWorker.hs index 6c12b02e816..318d35209fa 100644 --- a/services/background-worker/src/Wire/BackgroundWorker.hs +++ b/services/background-worker/src/Wire/BackgroundWorker.hs @@ -85,15 +85,14 @@ run opts galleyOpts = do withNamedLogger "migrate-users" $ Migrations.users opts.migrationOptions else pure $ pure () - - cleanupJobs <- - runAppT env $ - withNamedLogger "background-job-consumer" $ - Jobs.startWorker amqpEP cleanupJobRunner <- runAppT env $ withNamedLogger "job-runner" $ Workers.startWorker opts.jobs opts.meetingsCleanup + cleanupJobs <- + runAppT env $ + withNamedLogger "background-job-consumer" $ + Jobs.startWorker amqpEP let cleanup = void $ runConcurrently $ diff --git a/services/background-worker/src/Wire/BackgroundWorker/Jobs/Registry.hs b/services/background-worker/src/Wire/BackgroundWorker/Jobs/Registry.hs index abb737a95a0..62384214cb9 100644 --- a/services/background-worker/src/Wire/BackgroundWorker/Jobs/Registry.hs +++ b/services/background-worker/src/Wire/BackgroundWorker/Jobs/Registry.hs @@ -38,4 +38,4 @@ dispatchJob job = do $ runBackgroundWorkerEffects env extEnv job.requestId (Just job.jobId) . interpretBackgroundJobPublisherRabbitMQ job.requestId env.amqpJobsPublisherChannel . interpretBackgroundJobRunner - $ runJob job + $ runJob job.payload diff --git a/services/background-worker/src/Wire/BackgroundWorker/Workers.hs b/services/background-worker/src/Wire/BackgroundWorker/Workers.hs index e28c1e92d14..a8c150b92fd 100644 --- a/services/background-worker/src/Wire/BackgroundWorker/Workers.hs +++ b/services/background-worker/src/Wire/BackgroundWorker/Workers.hs @@ -22,10 +22,12 @@ module Wire.BackgroundWorker.Workers (startWorker) where import Arbiter.Core qualified as ArbiterCore +import Arbiter.Core.Exceptions (throwRetryable) import Arbiter.Worker qualified as ArbiterWorker import Arbiter.Worker.Config qualified as ArbiterWorkerConfig import Arbiter.Worker.Cron qualified as ArbiterWorkerCron import Control.Exception (throwIO) +import Data.Id (RequestId) import Data.Misc (Duration, duration) import Data.Range (fromRange) import Data.Text qualified as T @@ -35,12 +37,18 @@ import System.Cron (CronSchedule, serializeCronSchedule) import System.IO.Error (userError) import System.Logger qualified as Log import UnliftIO.Async qualified as Async +import Wire.API.BackgroundJobs (BackgroundJobPayload (..)) import Wire.API.Jobs import Wire.AdminlessJobsWorker (runAdminlessDeletionJob, runAdminlessReminderJob, runAdminlessSetupJob) +import Wire.BackgroundJobsPublisher.Arbiter (interpretBackgroundJobPublisherArbiter) +import Wire.BackgroundJobsRunner (runJob) +import Wire.BackgroundJobsRunner.Interpreter (interpretBackgroundJobRunner) import Wire.BackgroundWorker.Env (AppT, Env (..), runAppT) import Wire.BackgroundWorker.Options (JobConfig (..), JobJitter (..), MeetingsCleanupConfig (..)) import Wire.BackgroundWorker.Util +import Wire.Effects (runBackgroundWorkerEffects) import Wire.ExternalAccess.External +import Wire.JobSubsystem (JobSubsystemConfig (..)) import Wire.JobSubsystem.ArbiterAdapter import Wire.JobSubsystem.Migrations (runJobMigrations) import Wire.MeetingsCleanupWorker @@ -132,7 +140,7 @@ runJobRunner :: runJobRunner env extEnv runnerConfig cleanupConfig = do Log.info runnerConfig.jobRunnerLogger $ Log.msg (Log.val "Starting job worker") - . Log.field "queue_names" (T.intercalate "," [meetingsQueueName, conversationsQueueName]) + . Log.field "queue_names" (T.intercalate "," [meetingsQueueName, conversationsQueueName, userGroupsQueueName]) . Log.field "schedule" (show runnerConfig.jobRunnerSchedule) let arbiterEnv = mkNewWireArbiterEnv runnerConfig.jobRunnerSchemaName env.hasqlPool @@ -154,6 +162,19 @@ runJobRunner env extEnv runnerConfig cleanupConfig = do AdminlessDeletion payload -> runAppT env $ runAdminlessDeletionJob extEnv (mapJobPayload (const payload) job) AdminlessReminder payload -> runAppT env $ runAdminlessReminderJob extEnv (mapJobPayload (const payload) job) + userGroupsWorkerHandler _conn job = liftIO $ do + Log.info runnerConfig.jobRunnerLogger $ + Log.msg (Log.val "Running job") + . Log.field "queue_name" userGroupsQueueName + . Log.field "payload_type" (userGroupsJobPayloadTypeName job.payload) + case job.payload of + UserGroupsSyncUserGroup payload -> + runUserGroupsJob env extEnv runnerConfig.jobRunnerSchemaName payload.userGroupsSyncUserGroupJobRequestId $ + BackgroundJobSyncUserGroup payload.userGroupsSyncUserGroupJobData + UserGroupsSyncUserGroupAndChannel payload -> + runUserGroupsJob env extEnv runnerConfig.jobRunnerSchemaName payload.userGroupsSyncUserGroupAndChannelJobRequestId $ + BackgroundJobSyncUserGroupAndChannel payload.userGroupsSyncUserGroupAndChannelJobData + cronJob <- case ArbiterWorkerCron.cronJob "meetings-cleanup" (serializeCronSchedule runnerConfig.jobRunnerSchedule) @@ -189,6 +210,17 @@ runJobRunner env extEnv runnerConfig cleanupConfig = do ) ) + userGroupsWorkerConfig <- + ( ArbiterWorker.transactionalWorkerConfig + runnerConfig.jobRunnerSettings.jobWorkerThreads + userGroupsWorkerHandler :: + IO + ( ArbiterWorker.WorkerConfig + (WireArbiter JobRegistry) + UserGroupsJobPayload + ) + ) + let meetingsWorkerConfig' = applyExplicitDefaults runnerConfig.jobRunnerSettings @@ -201,7 +233,9 @@ runJobRunner env extEnv runnerConfig cleanupConfig = do conversationsWorkerConfig workerPools = [ ArbiterWorker.namedWorkerPool meetingsWorkerConfig', - ArbiterWorker.namedWorkerPool conversationsWorkerConfig' + ArbiterWorker.namedWorkerPool conversationsWorkerConfig', + ArbiterWorker.namedWorkerPool $ + applyExplicitDefaults runnerConfig.jobRunnerSettings userGroupsWorkerConfig ] workerAsync <- @@ -223,6 +257,11 @@ conversationsJobPayloadTypeName = \case AdminlessDeletion _ -> "adminless_deletion" AdminlessReminder _ -> "adminless_reminder" +userGroupsJobPayloadTypeName :: UserGroupsJobPayload -> Text +userGroupsJobPayloadTypeName = \case + UserGroupsSyncUserGroup _ -> "sync_user_group" + UserGroupsSyncUserGroupAndChannel _ -> "sync_user_group_and_channel" + mapJobPayload :: (a -> b) -> ArbiterCore.JobRead a -> ArbiterCore.JobRead b mapJobPayload f job = ArbiterCore.Job @@ -247,6 +286,18 @@ mapJobPayload f job = ArbiterCore.admission = job.admission } +-- | Run a user-groups sync job through the standard dispatch pipeline, +-- publishing any chained syncs back onto the Arbiter user-groups queue. +-- Failures are raised as retryable so Arbiter retries up to 'maxAttempts'. +runUserGroupsJob :: Env -> ExtEnv -> Text -> RequestId -> BackgroundJobPayload -> IO () +runUserGroupsJob env extEnv schemaName requestId payload = do + result <- + runBackgroundWorkerEffects env extEnv requestId Nothing + . interpretBackgroundJobPublisherArbiter requestId (JobSubsystemConfig schemaName) + . interpretBackgroundJobRunner + $ runJob payload + either throwRetryable pure result + applyExplicitDefaults :: JobWorkerSettings -> ArbiterWorker.WorkerConfig m payload -> diff --git a/services/brig/brig.cabal b/services/brig/brig.cabal index 794b137836c..acc88bea890 100644 --- a/services/brig/brig.cabal +++ b/services/brig/brig.cabal @@ -209,6 +209,7 @@ library , amazonka-ses >=2 , amazonka-sqs >=2 , amqp + , arbiter-core , async >=2.1 , auto-update >=0.1 , base >=4 && <5 diff --git a/services/brig/default.nix b/services/brig/default.nix index e431f9c93db..a556d7ad7cc 100644 --- a/services/brig/default.nix +++ b/services/brig/default.nix @@ -9,6 +9,7 @@ , amazonka-ses , amazonka-sqs , amqp +, arbiter-core , async , attoparsec , auto-update @@ -161,6 +162,7 @@ mkDerivation { amazonka-ses amazonka-sqs amqp + arbiter-core async auto-update base diff --git a/services/brig/src/Brig/App.hs b/services/brig/src/Brig/App.hs index 6c2145ea8cd..cb6c9d9b7f9 100644 --- a/services/brig/src/Brig/App.hs +++ b/services/brig/src/Brig/App.hs @@ -73,7 +73,6 @@ module Brig.App disabledVersionsLens, enableSFTFederationLens, rateLimitEnvLens, - amqpJobsPublisherChannelLens, postgresMigrationLens, initZAuth, initLogger, @@ -226,7 +225,6 @@ data Env = Env disabledVersions :: Set Version, enableSFTFederation :: Maybe Bool, rateLimitEnv :: RateLimitEnv, - amqpJobsPublisherChannel :: MVar Q.Channel, postgresMigration :: PostgresMigrationOpts } @@ -300,7 +298,6 @@ newEnv opts = do idxEnv <- mkIndexEnv opts.elasticsearch lgr (Opt.galley opts) mgr rateLimitEnv <- newRateLimitEnv opts.settings.passwordHashingRateLimit hasqlPool <- initPostgresPool opts.postgresqlPool opts.postgresql opts.postgresqlPassword - amqpJobsPublisherChannel <- Q.mkRabbitMqChannelMVar lgr (Just "brig") opts.rabbitmq pubKeyBundle <- loadPublicKeyBundle lgr opts.settings.publicKeyBundle pure $! Env @@ -344,7 +341,6 @@ newEnv opts = do disabledVersions = allDisabledVersions, enableSFTFederation = opts.multiSFT, rateLimitEnv, - amqpJobsPublisherChannel, postgresMigration = opts.postgresMigration } where diff --git a/services/brig/src/Brig/CanonicalInterpreter.hs b/services/brig/src/Brig/CanonicalInterpreter.hs index d44c91648d3..ea61f91d3ac 100644 --- a/services/brig/src/Brig/CanonicalInterpreter.hs +++ b/services/brig/src/Brig/CanonicalInterpreter.hs @@ -17,6 +17,7 @@ module Brig.CanonicalInterpreter where +import Arbiter.Core qualified as ArbiterCore import Brig.AWS (amazonkaEnv, prekeyTable) import Brig.App as App import Brig.DeleteQueue.Interpreter as DQ @@ -65,7 +66,7 @@ import Wire.BackendNotificationQueueAccess (BackendNotificationQueueAccess) import Wire.BackendNotificationQueueAccess.RabbitMq (interpretBackendNotificationQueueAccess) import Wire.BackendNotificationQueueAccess.RabbitMq qualified as BackendNotificationQueueAccess import Wire.BackgroundJobsPublisher (BackgroundJobPublisher) -import Wire.BackgroundJobsPublisher.RabbitMQ (interpretBackgroundJobPublisherRabbitMQ) +import Wire.BackgroundJobsPublisher.Arbiter (interpretBackgroundJobPublisherArbiter) import Wire.BlockListStore import Wire.BlockListStore.Cassandra import Wire.BudgetStore @@ -109,6 +110,7 @@ import Wire.IndexedUserStore import Wire.IndexedUserStore.ElasticSearch import Wire.InvitationStore (InvitationStore) import Wire.InvitationStore.Cassandra (interpretInvitationStoreToCassandra) +import Wire.JobSubsystem (JobSubsystemConfig (..)) import Wire.JwtTools import Wire.MigrationLock import Wire.MlsKeyPackageStore (MlsKeyPackageStore) @@ -498,7 +500,7 @@ runBrigToIO e (AppT ma) = do . interpretMlsKeyPackageSubsystem e.settings.keyPackageMaximumLifetime e.keyPackageLocalLock . interpretUserKeyStoreCassandra e.casClient . interpretRateLimit e.rateLimitEnv - . interpretBackgroundJobPublisherRabbitMQ e.requestId e.amqpJobsPublisherChannel + . interpretBackgroundJobPublisherArbiter e.requestId (JobSubsystemConfig ArbiterCore.defaultSchemaName) . interpretBackendNotificationQueueAccess (Just backendNotificationQueueEnv) . runNotificationSubsystemGundeck (defaultNotificationSubsystemConfig e.requestId) . runEvents diff --git a/services/brig/src/Brig/Run.hs b/services/brig/src/Brig/Run.hs index 20365d1ebb9..f14d542b2a9 100644 --- a/services/brig/src/Brig/Run.hs +++ b/services/brig/src/Brig/Run.hs @@ -18,6 +18,7 @@ module Brig.Run (run, mkApp, migratePostgres) where import AWS.Util (readAuthExpiration) +import Arbiter.Core qualified as ArbiterCore import Brig.API.Federation import Brig.API.Handler import Brig.API.Internal qualified as IAPI @@ -85,6 +86,8 @@ run :: Opts -> IO () run opts = withTracer \tracer -> do (app, e) <- mkApp opts runAllMigrations e.hasqlPool.rawPool e.appLogger + arbiterConnStr <- mkArbiterConnectionString opts.postgresql opts.postgresqlPassword + runJobMigrations arbiterConnStr ArbiterCore.defaultSchemaName let s = Server.newSettings (server e) internalEventListener <- Async.async $