From ed7252b34810e6140cdad886d2aafddd5cd9675f Mon Sep 17 00:00:00 2001 From: Gautier DI FOLCO Date: Fri, 28 Aug 2026 10:51:42 +0200 Subject: [PATCH 1/4] [WPB-28375] Move sync-user-group background jobs from RabbitMQ to Arbiter Add a 'user-groups' queue to the Arbiter job registry with a UserGroupsJobPayload sum (sync_user_group / sync_user_group_and_channel). brig now publishes sync jobs through the shared Postgres pool (per-group serialization, 3 attempts) and runs the registry migrations at startup. background-worker consumes the queue in a third Arbiter worker pool, reusing the existing dispatch pipeline; chained syncs are published back onto Arbiter and failures are retried via throwRetryable. The RabbitMQ jobs path is removed: jobs consumer and registry modules, RabbitMQ publisher interpreter, the BackgroundJob envelope and its dead serialization, the amqpJobsPublisherChannel in brig and background-worker, the backgroundJobs config block, and the corresponding charts, docs, and queue assertions in Test.Events. Backend-notification push and dead-user-notification watching remain on RabbitMQ. --- .../2-features/WPB-28375-user-groups-arbiter | 3 + charts/integration/templates/configmap.yaml | 6 - .../background-worker/configmap.yaml | 4 - charts/wire-server/values.yaml | 9 -- .../src/developer/reference/config-options.md | 23 +-- hack/helm_vars/wire-server/values.yaml.gotmpl | 4 - integration/test/Test/Events.hs | 7 +- libs/wire-api/src/Wire/API/BackgroundJobs.hs | 104 +------------ libs/wire-api/src/Wire/API/Jobs.hs | 98 +++++++++++- .../unit/Test/Wire/API/Roundtrip/Aeson.hs | 2 - .../src/Wire/BackgroundJobsPublisher.hs | 3 +- .../Wire/BackgroundJobsPublisher/Arbiter.hs | 82 ++++++++++ .../Wire/BackgroundJobsPublisher/RabbitMQ.hs | 66 -------- .../src/Wire/BackgroundJobsRunner.hs | 4 +- .../Wire/BackgroundJobsRunner/Interpreter.hs | 11 +- .../Wire/UserGroupSubsystem/Interpreter.hs | 34 ++-- libs/wire-subsystems/wire-subsystems.cabal | 2 +- .../background-worker/background-worker.cabal | 4 - .../background-worker.integration.yaml | 6 - services/background-worker/default.nix | 4 - .../src/Wire/BackgroundWorker.hs | 7 - .../src/Wire/BackgroundWorker/Env.hs | 14 +- .../Wire/BackgroundWorker/Jobs/Consumer.hs | 147 ------------------ .../Wire/BackgroundWorker/Jobs/Registry.hs | 41 ----- .../src/Wire/BackgroundWorker/Options.hs | 14 +- .../src/Wire/BackgroundWorker/Workers.hs | 55 ++++++- .../Wire/BackendNotificationPusherSpec.hs | 14 -- .../background-worker/test/Test/Wire/Util.hs | 9 -- services/brig/brig.cabal | 1 + services/brig/default.nix | 2 + services/brig/src/Brig/App.hs | 4 - .../brig/src/Brig/CanonicalInterpreter.hs | 6 +- services/brig/src/Brig/Run.hs | 3 + 33 files changed, 276 insertions(+), 517 deletions(-) create mode 100644 changelog.d/2-features/WPB-28375-user-groups-arbiter create mode 100644 libs/wire-subsystems/src/Wire/BackgroundJobsPublisher/Arbiter.hs delete mode 100644 libs/wire-subsystems/src/Wire/BackgroundJobsPublisher/RabbitMQ.hs delete mode 100644 services/background-worker/src/Wire/BackgroundWorker/Jobs/Consumer.hs delete mode 100644 services/background-worker/src/Wire/BackgroundWorker/Jobs/Registry.hs 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..192e6631f97 --- /dev/null +++ b/changelog.d/2-features/WPB-28375-user-groups-arbiter @@ -0,0 +1,3 @@ +Move the sync-user-group family of background jobs from the RabbitMQ `background-jobs` queue to the Arbiter Postgres 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 consumer, publisher channel, and `backgroundJobs` configuration are removed; backend-notification push and dead-user-notification watching remain on RabbitMQ. + +Rollout note for operators: deploy only after the RabbitMQ `background-jobs` queue has drained to zero. Jobs still in that queue when the last old background-worker pod terminates are dropped; sync jobs are idempotent reconciliations re-triggered by the next SCIM write, so bounded loss is acceptable. diff --git a/charts/integration/templates/configmap.yaml b/charts/integration/templates/configmap.yaml index 02a9d4db410..04e4bd16f11 100644 --- a/charts/integration/templates/configmap.yaml +++ b/charts/integration/templates/configmap.yaml @@ -56,12 +56,6 @@ data: backgroundWorker: host: background-worker.{{ .Release.Namespace }}.svc.cluster.local port: 8080 - # Background jobs defaults for integration tests - backgroundJobs: - concurrency: 4 - jobTimeout: 5s - maxAttempts: 3 - stern: host: stern.{{ .Release.Namespace }}.svc.cluster.local port: 8080 diff --git a/charts/wire-server/templates/background-worker/configmap.yaml b/charts/wire-server/templates/background-worker/configmap.yaml index 299d0703d3a..34a644af6cf 100644 --- a/charts/wire-server/templates/background-worker/configmap.yaml +++ b/charts/wire-server/templates/background-worker/configmap.yaml @@ -90,10 +90,6 @@ data: backendNotificationPusher: {{ toYaml .backendNotificationPusher | indent 6 }} - {{- with .backgroundJobs }} - backgroundJobs: -{{ toYaml . | indent 6 }} - {{- end }} jobs: # Arbiter dispatcher poll interval for all jobs. # Lower values reduce discovery latency, but increase DB polling. diff --git a/charts/wire-server/values.yaml b/charts/wire-server/values.yaml index c6dda2176a2..178a07dd372 100644 --- a/charts/wire-server/values.yaml +++ b/charts/wire-server/values.yaml @@ -1019,15 +1019,6 @@ background-worker: pushBackoffMaxWait: 300000000 # microseconds, so 300s remotesRefreshInterval: 300000000 # microseconds, so 300s - # Background jobs consumer configuration - backgroundJobs: - # Maximum number of in-flight jobs per process - concurrency: 8 - # Per-attempt timeout in seconds - jobTimeout: 60s - # Total attempts, including the first try - maxAttempts: 3 - # Job dispatcher configuration. jobs: # Arbiter dispatcher poll interval for all jobs. diff --git a/docs/src/developer/reference/config-options.md b/docs/src/developer/reference/config-options.md index c1b48558c7c..a41ef3ab30f 100644 --- a/docs/src/developer/reference/config-options.md +++ b/docs/src/developer/reference/config-options.md @@ -2282,12 +2282,6 @@ migrationOptions: # Required migration timeout for a single migration attempt. timeout: 5s -# Background jobs consumer -backgroundJobs: - concurrency: 8 # in-flight jobs per process - jobTimeout: 60s # per attempt - maxAttempts: 3 # total attempts incl. first run - # Jobs jobs: pollInterval: 5s # how often due jobs are discovered @@ -2322,15 +2316,11 @@ 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 job queues (meetings, +conversations, and user groups), covering immediate and scheduled or recurring +jobs, and controls their dispatcher, worker-pool, visibility, retry, and +reaper behavior. User-group synchronization jobs run on the `user-groups` +queue; jobs for the same user group are serialized via their group key. # Required for addressing local vs remote backends federationDomain: example.org @@ -2367,10 +2357,9 @@ 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. - `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/hack/helm_vars/wire-server/values.yaml.gotmpl b/hack/helm_vars/wire-server/values.yaml.gotmpl index 673cb6da817..14832a6f253 100644 --- a/hack/helm_vars/wire-server/values.yaml.gotmpl +++ b/hack/helm_vars/wire-server/values.yaml.gotmpl @@ -650,10 +650,6 @@ background-worker: port: "5432" user: wire-server dbname: wire-server - backgroundJobs: - concurrency: 8 - jobTimeout: 60s - maxAttempts: 3 jobs: pollInterval: 1s # Poll every second so due jobs are discovered promptly in tests workerThreads: 1 diff --git a/integration/test/Test/Events.hs b/integration/test/Test/Events.hs index a172cf2ff0f..b060cbcf067 100644 --- a/integration/test/Test/Events.hs +++ b/integration/test/Test/Events.hs @@ -176,12 +176,11 @@ testTemporaryQueuesAreDeletedAfterUse = do aliceClientQueue = Queue {name = fromString aliceClientQueueName, vhost = fromString beResource.berVHost} deadNotifsQueue = Queue {name = fromString "dead-user-notifications", vhost = fromString beResource.berVHost} cellsEventsQueue = Queue {name = fromString "cells_events", vhost = fromString beResource.berVHost} - backgroundJobsQueue = Queue {name = fromString "background-jobs", vhost = fromString beResource.berVHost} -- Wait for queue for the new client to be created eventually $ do queuesBeforeWS <- rabbitmqAdmin.listQueuesByVHost (fromString beResource.berVHost) (fromString "") True 100 1 - queuesBeforeWS.items `shouldMatchSet` [deadNotifsQueue, cellsEventsQueue, aliceClientQueue, backgroundJobsQueue] + queuesBeforeWS.items `shouldMatchSet` [deadNotifsQueue, cellsEventsQueue, aliceClientQueue] runCodensity (createEventsWebSocket alice Nothing) $ \ws -> do handle <- randomHandle @@ -189,7 +188,7 @@ testTemporaryQueuesAreDeletedAfterUse = do queuesDuringWS <- rabbitmqAdmin.listQueuesByVHost (fromString beResource.berVHost) (fromString "") True 100 1 addJSONToFailureContext "queuesDuringWS" queuesDuringWS $ do - length queuesDuringWS.items `shouldMatchInt` 5 + length queuesDuringWS.items `shouldMatchInt` 4 -- We cannot use 'assertEvent' here because there is a race between the temp -- queue being created and rabbitmq fanning out the previous events. @@ -203,7 +202,7 @@ testTemporaryQueuesAreDeletedAfterUse = do eventually $ do queuesAfterWS <- rabbitmqAdmin.listQueuesByVHost (fromString beResource.berVHost) (fromString "") True 100 1 - queuesAfterWS.items `shouldMatchSet` [deadNotifsQueue, cellsEventsQueue, aliceClientQueue, backgroundJobsQueue] + queuesAfterWS.items `shouldMatchSet` [deadNotifsQueue, cellsEventsQueue, aliceClientQueue] testSendMessageNoReturnToSenderWithConsumableNotificationsProteus :: (HasCallStack) => App () testSendMessageNoReturnToSenderWithConsumableNotificationsProteus = withModifiedBackend (enableConsumableNotifications def) $ \domain -> do diff --git a/libs/wire-api/src/Wire/API/BackgroundJobs.hs b/libs/wire-api/src/Wire/API/BackgroundJobs.hs index b96f2cd7505..772073e2be0 100644 --- a/libs/wire-api/src/Wire/API/BackgroundJobs.hs +++ b/libs/wire-api/src/Wire/API/BackgroundJobs.hs @@ -1,5 +1,4 @@ {-# LANGUAGE StrictData #-} -{-# LANGUAGE TemplateHaskell #-} -- This file is part of the Wire Server implementation. -- @@ -20,16 +19,11 @@ module Wire.API.BackgroundJobs where -import Control.Arrow ((&&&)) -import Control.Lens (makePrisms) import Data.Aeson qualified as Aeson import Data.Id -import Data.Map.Strict qualified as Map import Data.OpenApi qualified as S import Data.Schema import Imports -import Network.AMQP qualified as Q -import Network.AMQP.Types qualified as QT import Wire.Arbitrary (Arbitrary (..), GenericUniform (..)) data BackgroundJobPayload @@ -38,34 +32,6 @@ data BackgroundJobPayload deriving stock (Eq, Show, Generic) deriving (Arbitrary) via GenericUniform BackgroundJobPayload -backgroundJobPayloadLabel :: BackgroundJobPayload -> Text -backgroundJobPayloadLabel p = case backgroundJobPayloadTag p of - BackgroundJobSyncUserGroupAndChannelTag -> "sync-user-group-and-channel" - BackgroundJobSyncUserGroupTag -> "sync-user-group" - -data BackgroundJobPayloadTag - = BackgroundJobSyncUserGroupAndChannelTag - | BackgroundJobSyncUserGroupTag - deriving stock (Eq, Ord, Bounded, Enum, Show, Generic) - deriving (Arbitrary) via GenericUniform BackgroundJobPayloadTag - -instance ToSchema BackgroundJobPayloadTag where - schema = - enum @Text $ - mconcat - [ element "sync-user-group-and-channel" BackgroundJobSyncUserGroupAndChannelTag, - element "sync-user-group" BackgroundJobSyncUserGroupTag - ] - -backgroundJobPayloadTag :: BackgroundJobPayload -> BackgroundJobPayloadTag -backgroundJobPayloadTag = - \case - BackgroundJobSyncUserGroupAndChannel {} -> BackgroundJobSyncUserGroupAndChannelTag - BackgroundJobSyncUserGroup {} -> BackgroundJobSyncUserGroupTag - -backgroundJobPayloadTagSchema :: ObjectSchema SwaggerDoc BackgroundJobPayloadTag -backgroundJobPayloadTagSchema = field "type" schema - data SyncUserGroupAndChannel = SyncUserGroupAndChannel { teamId :: TeamId, userGroupId :: UserGroupId, @@ -73,7 +39,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 +57,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 @@ -101,69 +67,3 @@ instance ToSchema SyncUserGroup where <$> (.teamId) .= field "team_id" schema <*> (.userGroupId) .= field "user_group_id" schema <*> (.actor) .= maybe_ (optField "actor" schema) - -makePrisms ''BackgroundJobPayload - -backgroundJobPayloadObjectSchema :: ObjectSchema SwaggerDoc BackgroundJobPayload -backgroundJobPayloadObjectSchema = - snd - <$> (backgroundJobPayloadTag &&& id) - .= bind - (fst .= backgroundJobPayloadTagSchema) - (snd .= dispatch backgroundJobPayloadDataSchema) - where - backgroundJobPayloadDataSchema :: BackgroundJobPayloadTag -> ObjectSchema SwaggerDoc BackgroundJobPayload - backgroundJobPayloadDataSchema = \case - BackgroundJobSyncUserGroupAndChannelTag -> tag _BackgroundJobSyncUserGroupAndChannel (field "payload" schema) - BackgroundJobSyncUserGroupTag -> tag _BackgroundJobSyncUserGroup (field "payload" schema) - -instance ToSchema BackgroundJobPayload where - schema = object backgroundJobPayloadObjectSchema - -deriving via (Schema BackgroundJobPayload) instance Aeson.FromJSON BackgroundJobPayload - -deriving via (Schema BackgroundJobPayload) instance Aeson.ToJSON BackgroundJobPayload - -deriving via (Schema BackgroundJobPayload) instance S.ToSchema BackgroundJobPayload - --- | Background job envelope. Payload is a free-form JSON object. -data BackgroundJob = BackgroundJob - { jobId :: JobId, - requestId :: RequestId, - payload :: BackgroundJobPayload - } - deriving stock (Eq, Show, Generic) - deriving (Arbitrary) via GenericUniform BackgroundJob - deriving (Aeson.ToJSON, Aeson.FromJSON, S.ToSchema) via Schema BackgroundJob - -instance ToSchema BackgroundJob where - schema = - object $ - BackgroundJob - <$> jobId .= field "id" schema - <*> requestId .= field "requestId" schema - <*> payload .= field "payload" schema - -backgroundJobsRoutingKey :: Text -backgroundJobsRoutingKey = backgroundJobsQueueName - -backgroundJobsQueueName :: Text -backgroundJobsQueueName = "background-jobs" - -ensureBackgroundJobsQueue :: Q.Channel -> IO () -ensureBackgroundJobsQueue chan = do - let headers = - QT.FieldTable - ( Map.fromList - [ ("x-queue-type", QT.FVString "quorum") - ] - ) - q = - Q.newQueue - { Q.queueName = backgroundJobsQueueName, - Q.queueDurable = True, - Q.queueAutoDelete = False, - Q.queueExclusive = False, - Q.queueHeaders = headers - } - void $ Q.declareQueue chan q 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-api/test/unit/Test/Wire/API/Roundtrip/Aeson.hs b/libs/wire-api/test/unit/Test/Wire/API/Roundtrip/Aeson.hs index 14f6f83c610..0c2a791f0a6 100644 --- a/libs/wire-api/test/unit/Test/Wire/API/Roundtrip/Aeson.hs +++ b/libs/wire-api/test/unit/Test/Wire/API/Roundtrip/Aeson.hs @@ -31,7 +31,6 @@ import Test.Tasty.HUnit (assertEqual, assertFailure, testCase) import Test.Tasty.QuickCheck (Arbitrary, counterexample, testProperty, (.&&.), (===)) import Type.Reflection (typeRep) import Wire.API.Asset qualified as Asset -import Wire.API.BackgroundJobs qualified as BackgroundJobs import Wire.API.Call.Config qualified as Call.Config import Wire.API.Connection qualified as Connection import Wire.API.Conversation qualified as Conversation @@ -384,7 +383,6 @@ tests = testRoundTrip @TeamsIntra.TeamStatusUpdate, testRoundTrip @TeamsIntra.TeamData, testRoundTrip @TeamsIntra.TeamName, - testRoundTrip @BackgroundJobs.BackgroundJob, testRoundTrip @User.ManagedByUpdate, testRoundTrip @User.Auth.ReAuthUser, testRoundTrip @User.RichInfoUpdate, 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 deleted file mode 100644 index 1b3e9cb273b..00000000000 --- a/libs/wire-subsystems/src/Wire/BackgroundJobsPublisher/RabbitMQ.hs +++ /dev/null @@ -1,66 +0,0 @@ --- This file is part of the Wire Server implementation. --- --- Copyright (C) 2025 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.RabbitMQ where - -import Data.Aeson qualified as Aeson -import Data.Id (JobId, RequestId (..), idToText) -import Data.Text.Encoding qualified as T -import Imports -import Network.AMQP qualified as Q -import Polysemy -import Wire.API.BackgroundJobs -import Wire.BackgroundJobsPublisher (BackgroundJobPublisher (..)) - -interpretBackgroundJobPublisherRabbitMQ :: - (Member (Embed IO) r) => - RequestId -> - MVar Q.Channel -> - InterpreterFor BackgroundJobPublisher r -interpretBackgroundJobPublisherRabbitMQ requestId channelMVar = - interpret $ \case - PublishJob jobId jobPayload -> do - channel <- readMVar channelMVar - publishJob requestId channel jobId jobPayload - -publishJob :: - (Member (Embed IO) r) => - RequestId -> - Q.Channel -> - JobId -> - BackgroundJobPayload -> - Sem r () -publishJob requestId channel jobId jobPayload = do - let job = - BackgroundJob - { payload = jobPayload, - jobId = jobId, - requestId = requestId - } - msg = - Q.newMsg - { Q.msgBody = Aeson.encode job, - Q.msgContentType = Just "application/json", - Q.msgID = Just (idToText job.jobId), - Q.msgCorrelationID = Just $ T.decodeUtf8 job.requestId.unRequestId - } - - liftIO $ do - ensureBackgroundJobsQueue channel - -- Passing "" for `exchangeName` publishes to the default exchange in RabbitMQ. - -- The default exchange routes directly to the queue whose name equals the `routingKey`. - void $ Q.publishMsg channel "" backgroundJobsRoutingKey msg 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..07b2d84c4e5 100644 --- a/libs/wire-subsystems/wire-subsystems.cabal +++ b/libs/wire-subsystems/wire-subsystems.cabal @@ -236,7 +236,7 @@ library Wire.BackendNotificationQueueAccess Wire.BackendNotificationQueueAccess.RabbitMq Wire.BackgroundJobsPublisher - Wire.BackgroundJobsPublisher.RabbitMQ + Wire.BackgroundJobsPublisher.Arbiter Wire.BackgroundJobsRunner Wire.BackgroundJobsRunner.Interpreter Wire.BlockListStore diff --git a/services/background-worker/background-worker.cabal b/services/background-worker/background-worker.cabal index a43796a4ca7..ccf77722432 100644 --- a/services/background-worker/background-worker.cabal +++ b/services/background-worker/background-worker.cabal @@ -17,8 +17,6 @@ library Wire.BackgroundWorker Wire.BackgroundWorker.Env Wire.BackgroundWorker.Health - Wire.BackgroundWorker.Jobs.Consumer - Wire.BackgroundWorker.Jobs.Registry Wire.BackgroundWorker.Options Wire.BackgroundWorker.Util Wire.BackgroundWorker.Workers @@ -46,10 +44,8 @@ library , cassandra-util , containers , cron - , data-timeout , exceptions , extended - , extra , galley-types , hasql-resource-pool , HsOpenSSL diff --git a/services/background-worker/background-worker.integration.yaml b/services/background-worker/background-worker.integration.yaml index b0bd0d172e4..9b89ac94b12 100644 --- a/services/background-worker/background-worker.integration.yaml +++ b/services/background-worker/background-worker.integration.yaml @@ -60,12 +60,6 @@ migrateTeamFeatures: false migrateDomainRegistration: false migrateUsers: false -# Background jobs consumer configuration for integration -backgroundJobs: - concurrency: 4 - jobTimeout: 5s - maxAttempts: 3 - # Job dispatcher configuration for integration jobs: pollInterval: 1s # Poll every second so due jobs are discovered promptly in tests diff --git a/services/background-worker/default.nix b/services/background-worker/default.nix index acf79b6e195..06ff0a24ebb 100644 --- a/services/background-worker/default.nix +++ b/services/background-worker/default.nix @@ -15,10 +15,8 @@ , containers , cron , data-default -, data-timeout , exceptions , extended -, extra , federator , galley-types , hasql-resource-pool @@ -77,10 +75,8 @@ mkDerivation { cassandra-util containers cron - data-timeout exceptions extended - extra galley-types hasql-resource-pool HsOpenSSL diff --git a/services/background-worker/src/Wire/BackgroundWorker.hs b/services/background-worker/src/Wire/BackgroundWorker.hs index 6c12b02e816..c2c32df5e02 100644 --- a/services/background-worker/src/Wire/BackgroundWorker.hs +++ b/services/background-worker/src/Wire/BackgroundWorker.hs @@ -31,7 +31,6 @@ import Util.Options import Wire.BackendNotificationPusher qualified as BackendNotificationPusher import Wire.BackgroundWorker.Env import Wire.BackgroundWorker.Health qualified as Health -import Wire.BackgroundWorker.Jobs.Consumer qualified as Jobs import Wire.BackgroundWorker.Options import Wire.BackgroundWorker.Workers as Workers import Wire.DeadUserNotificationWatcher qualified as DeadUserNotificationWatcher @@ -85,11 +84,6 @@ 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" $ @@ -106,7 +100,6 @@ run opts galleyOpts = do <*> Concurrently cleanupDomainRegistrationMigration <*> Concurrently cleanupUsersMigration <*> Concurrently cleanupJobRunner - <*> Concurrently cleanupJobs let server = defaultServer (T.unpack opts.backgroundWorker.host) opts.backgroundWorker.port env.logger let settings = newSettings server diff --git a/services/background-worker/src/Wire/BackgroundWorker/Env.hs b/services/background-worker/src/Wire/BackgroundWorker/Env.hs index ed784a33db9..2fe576c57ed 100644 --- a/services/background-worker/src/Wire/BackgroundWorker/Env.hs +++ b/services/background-worker/src/Wire/BackgroundWorker/Env.hs @@ -64,14 +64,12 @@ type IsWorking = Bool data Worker = BackendNotificationPusher | DeadUserNotificationWatcher - | BackgroundJobConsumer deriving (Eq, Ord) workerName :: Worker -> Text workerName = \case BackendNotificationPusher -> "backend-notification-pusher" DeadUserNotificationWatcher -> "dead-user-notification-watcher" - BackgroundJobConsumer -> "background-job-consumer" data Env = Env { http2Manager :: Http2Manager, @@ -84,7 +82,6 @@ data Env = Env backendNotificationMetrics :: BackendNotificationMetrics, meetingsCleanupMetrics :: MeetingsCleanupMetrics, backendNotificationsConfig :: BackendNotificationsConfig, - backgroundJobsConfig :: BackgroundJobsConfig, workerRunningGauge :: Vector Text Gauge, statuses :: IORef (Map Worker IsWorking), cassandra :: ClientState, @@ -93,8 +90,7 @@ data Env = Env hasqlPool :: Hasql.Pool, -- May contain the PostgreSQL password. Do not unwrap outside the Arbiter boundary. arbiterConnStr :: SecretText, - -- Dedicated AMQP channels per concern - amqpJobsPublisherChannel :: MVar Q.Channel, + -- Dedicated AMQP channel per concern amqpBackendNotificationsChannel :: MVar Q.Channel, federationDomain :: Domain, postgresMigration :: PostgresMigrationOpts, @@ -170,13 +166,11 @@ mkEnv opts galleyOpts = do statuses <- newIORef $ Map.fromList - [ (BackendNotificationPusher, False), - (BackgroundJobConsumer, False) + [ (BackendNotificationPusher, False) ] backendNotificationMetrics <- mkBackendNotificationMetrics meetingsCleanupMetrics <- mkMeetingsCleanupMetrics let backendNotificationsConfig = opts.backendNotificationPusher - backgroundJobsConfig = opts.backgroundJobs federationDomain = galleyOpts._settings._federationDomain postgresMigration = opts.postgresMigration brigEndpoint = opts.brig @@ -195,10 +189,6 @@ mkEnv opts galleyOpts = do workerRunningGauge <- mkWorkerRunningGauge hasqlPool <- initPostgresPool opts.postgresqlPool galleyOpts._postgresql galleyOpts._postgresqlPassword arbiterConnStr <- mkArbiterConnectionString galleyOpts._postgresql galleyOpts._postgresqlPassword - Log.info logger $ Log.msg @Text "Opening RabbitMQ channel: background-worker-jobs-publisher..." - amqpJobsPublisherChannel <- - mkRabbitMqChannelMVar logger (Just "background-worker-jobs-publisher") $ - either id demoteOpts opts.rabbitmq.unRabbitMqOpts Log.info logger $ Log.msg @Text "Opening RabbitMQ channel: background-worker-backend-notifications..." amqpBackendNotificationsChannel <- mkRabbitMqChannelMVar logger (Just "background-worker-backend-notifications") $ diff --git a/services/background-worker/src/Wire/BackgroundWorker/Jobs/Consumer.hs b/services/background-worker/src/Wire/BackgroundWorker/Jobs/Consumer.hs deleted file mode 100644 index cb0d8a4b9c6..00000000000 --- a/services/background-worker/src/Wire/BackgroundWorker/Jobs/Consumer.hs +++ /dev/null @@ -1,147 +0,0 @@ -{-# LANGUAGE RecordWildCards #-} - --- This file is part of the Wire Server implementation. --- --- Copyright (C) 2025 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.BackgroundWorker.Jobs.Consumer (startWorker, BackgroundJobMetrics (..)) where - -import Control.Concurrent.Timeout qualified as Timeout -import Control.Retry -import Data.Aeson qualified as Aeson -import Data.Misc hiding (duration) -import Data.Range (Range (fromRange)) -import Data.Timeout -import Imports -import Network.AMQP qualified as Q -import Network.AMQP.Extended -import Network.AMQP.Lifted qualified as QL -import Prometheus -import System.Logger.Class qualified as Log -import System.Time.Extra (duration) -import UnliftIO -import Wire.API.BackgroundJobs -import Wire.BackgroundWorker.Env -import Wire.BackgroundWorker.Jobs.Registry -import Wire.BackgroundWorker.Options -import Wire.BackgroundWorker.Util (CleanupAction) - -data BackgroundJobMetrics = BackgroundJobMetrics - { workersBusy :: Gauge, - concurrencyConfigured :: Gauge, - jobsReceived :: Vector Text Counter, - jobsStarted :: Vector Text Counter, - jobsSucceeded :: Vector Text Counter, - jobsFailed :: Vector Text Counter, - jobsInvalid :: Vector Text Counter, - jobsRedelivered :: Vector Text Counter, - jobDuration :: Vector Text Histogram - } - -mkMetrics :: IO BackgroundJobMetrics -mkMetrics = do - workersBusy <- register (gauge $ Info {metricName = "wire_background_jobs_workers_busy", metricHelp = "In-flight background jobs"}) - concurrencyConfigured <- register (gauge $ Info {metricName = "wire_background_jobs_concurrency_configured", metricHelp = "Configured concurrency for this process"}) - jobsReceived <- register (vector "job_type" $ counter $ Info "wire_background_jobs_received_total" "Jobs received") - jobsStarted <- register (vector "job_type" $ counter $ Info "wire_background_jobs_started_total" "Jobs started") - jobsSucceeded <- register (vector "job_type" $ counter $ Info "wire_background_jobs_succeeded_total" "Jobs succeeded") - jobsFailed <- register (vector "job_type" $ counter $ Info "wire_background_jobs_failed_total" "Jobs failed") - jobsInvalid <- register (vector "job_type" $ counter $ Info "wire_background_jobs_invalid_total" "Invalid jobs received") - jobsRedelivered <- register (vector "job_type" $ counter $ Info "wire_background_jobs_redelivered_total" "Jobs marked redelivered by broker") - jobDuration <- register (vector "job_type" $ histogram (Info "wire_background_jobs_duration_seconds" "Job duration seconds") defaultBuckets) - pure BackgroundJobMetrics {..} - -startWorker :: AmqpEndpoint -> AppT IO CleanupAction -startWorker rabbitmqOpts = do - env <- ask - let cfg = env.backgroundJobsConfig - metrics <- liftIO mkMetrics - markAsNotWorking BackgroundJobConsumer - void . async . liftIO $ - openConnectionWithRetries env.logger rabbitmqOpts (Just "background-job-consumer") $ - RabbitMqHooks - { onNewChannel = \chan -> do - -- declare queue and set prefetch to concurrency - ensureBackgroundJobsQueue chan - Q.qos chan 0 (fromIntegral $ fromRange cfg.concurrency) False - -- set gauges - setGauge metrics.concurrencyConfigured (fromIntegral $ fromRange cfg.concurrency) - -- start consuming with manual ack and keep the channel alive - void $ QL.consumeMsgs chan backgroundJobsQueueName Q.Ack (void . runAppT env . handleDelivery metrics cfg) - runAppT env $ markAsWorking BackgroundJobConsumer - forever $ threadDelay maxBound, - onChannelException = \e -> do - runAppT env $ markAsNotWorking BackgroundJobConsumer - let connClosed = - case (fromException e :: Maybe Q.AMQPException) of - Just (Q.ConnectionClosedException _ _) -> True - _ -> False - unless (Q.isNormalChannelClose e || connClosed) $ - runAppT env $ - Log.err $ - Log.msg (Log.val "Unexpected RabbitMQ channel exception in background job consumer") - . Log.field "exception" (displayException e), - onConnectionClose = - runAppT env $ do - markAsNotWorking BackgroundJobConsumer - Log.info $ Log.msg (Log.val "RabbitMQ connection closed for background job consumer") - } - pure $ runAppT env $ cleanup - where - cleanup :: AppT IO () - cleanup = do - -- nothing to close explicitly; the AMQP helper closes channel/connection on shutdown - Log.info $ Log.msg (Log.val "Background job consumer cleanup") - markAsNotWorking BackgroundJobConsumer - -handleDelivery :: BackgroundJobMetrics -> BackgroundJobsConfig -> (Q.Message, Q.Envelope) -> AppT IO () -handleDelivery metrics cfg (msg, env) = do - case Aeson.eitherDecode @BackgroundJob (Q.msgBody msg) of - Left err -> do - withLabel metrics.jobsInvalid "invalid" incCounter - Log.err $ Log.msg (Log.val "Invalid background job JSON") . Log.field "error" err - Timeout.threadDelay (200 # MilliSecond) -- avoid tight redelivery loop - liftIO $ Q.rejectEnv env True - Right job -> do - let lbl = backgroundJobPayloadLabel job.payload - when (Q.envRedelivered env) $ withLabel metrics.jobsRedelivered lbl incCounter - withLabel metrics.jobsReceived lbl incCounter - UnliftIO.bracket_ (incGauge metrics.workersBusy) (decGauge metrics.workersBusy) $ do - outcome <- runAttempts lbl job - case outcome of - Right () -> do - withLabel metrics.jobsSucceeded lbl incCounter - liftIO $ Q.ackEnv env - Left e -> do - withLabel metrics.jobsFailed lbl incCounter - Log.err $ Log.msg (Log.val "Background job failed after retries") . Log.field "error" e - liftIO $ Q.rejectEnv env False - where - runAttempts :: Text -> BackgroundJob -> AppT IO (Either Text ()) - runAttempts lbl job = do - let retries = max 0 (fromRange cfg.maxAttempts - 1) - policy = limitRetries retries <> fullJitterBackoff 100000 -- 100ms base - retrying policy shouldRetry $ \_rs -> do - withLabel metrics.jobsStarted lbl incCounter - (dur, r) <- - duration $ - fromMaybe (Left "job timeout") - <$> timeout (durationToMicros cfg.jobTimeout) (dispatchJob job) - withLabel metrics.jobDuration lbl (`observe` dur) - pure r - where - shouldRetry _ (Right _) = pure False - shouldRetry _ (Left _) = pure True diff --git a/services/background-worker/src/Wire/BackgroundWorker/Jobs/Registry.hs b/services/background-worker/src/Wire/BackgroundWorker/Jobs/Registry.hs deleted file mode 100644 index abb737a95a0..00000000000 --- a/services/background-worker/src/Wire/BackgroundWorker/Jobs/Registry.hs +++ /dev/null @@ -1,41 +0,0 @@ --- This file is part of the Wire Server implementation. --- --- Copyright (C) 2025 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.BackgroundWorker.Jobs.Registry - ( dispatchJob, - ) -where - -import Imports -import Wire.API.BackgroundJobs (BackgroundJob (..)) -import Wire.BackgroundJobsPublisher.RabbitMQ (interpretBackgroundJobPublisherRabbitMQ) -import Wire.BackgroundJobsRunner (runJob) -import Wire.BackgroundJobsRunner.Interpreter (interpretBackgroundJobRunner) -import Wire.BackgroundWorker.Env (AppT, Env (..)) -import Wire.Effects -import Wire.ExternalAccess.External - -dispatchJob :: BackgroundJob -> AppT IO (Either Text ()) -dispatchJob job = do - env <- ask @Env - let disableTlsV1 = True - extEnv <- liftIO (initExtEnv disableTlsV1) - liftIO - $ runBackgroundWorkerEffects env extEnv job.requestId (Just job.jobId) - . interpretBackgroundJobPublisherRabbitMQ job.requestId env.amqpJobsPublisherChannel - . interpretBackgroundJobRunner - $ runJob job diff --git a/services/background-worker/src/Wire/BackgroundWorker/Options.hs b/services/background-worker/src/Wire/BackgroundWorker/Options.hs index 035460cfc32..0e9891507cf 100644 --- a/services/background-worker/src/Wire/BackgroundWorker/Options.hs +++ b/services/background-worker/src/Wire/BackgroundWorker/Options.hs @@ -57,8 +57,7 @@ data Opts = Opts migrateDomainRegistration :: !Bool, migrateUsers :: !Bool, jobs :: JobConfig, - meetingsCleanup :: MeetingsCleanupConfig, - backgroundJobs :: BackgroundJobsConfig + meetingsCleanup :: MeetingsCleanupConfig } deriving (Show, Generic) deriving (FromJSON) via Generically Opts @@ -90,17 +89,6 @@ instance FromJSON RabbitMqOpts where <|> (Left <$> parseJSON v) ) -data BackgroundJobsConfig = BackgroundJobsConfig - { -- | Maximum parallel jobs processed by this process - concurrency :: Range 1 1000 Int, - -- | Per-attempt timeout (seconds) - jobTimeout :: Duration, - -- | Total attempts including first run - maxAttempts :: Range 1 1000 Int - } - deriving (Show, Generic) - deriving (FromJSON) via Generically BackgroundJobsConfig - data JobConfig = JobConfig { -- | Arbiter dispatcher poll interval for jobs. -- Lower values reduce discovery latency for due jobs. 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/background-worker/test/Test/Wire/BackendNotificationPusherSpec.hs b/services/background-worker/test/Test/Wire/BackendNotificationPusherSpec.hs index 7222120d93a..66befeb0a03 100644 --- a/services/background-worker/test/Test/Wire/BackendNotificationPusherSpec.hs +++ b/services/background-worker/test/Test/Wire/BackendNotificationPusherSpec.hs @@ -355,14 +355,7 @@ spec = do rabbitmqVHost = "test-vhost" defederationTimeout = responseTimeoutNone backendNotificationsConfig = BackendNotificationsConfig 1000 500000 1000 - backgroundJobsConfig = - BackgroundJobsConfig - { concurrency = toRange (Proxy @1), - jobTimeout = Duration 100, - maxAttempts = toRange (Proxy @3) - } hasqlPool = undefined - amqpJobsPublisherChannel = undefined amqpBackendNotificationsChannel = undefined federationDomain = Domain "local" postgresMigration = @@ -419,14 +412,7 @@ spec = do rabbitmqVHost = "test-vhost" defederationTimeout = responseTimeoutNone backendNotificationsConfig = BackendNotificationsConfig 1000 500000 1000 - backgroundJobsConfig = - BackgroundJobsConfig - { concurrency = toRange (Proxy @1), - jobTimeout = Duration 100, - maxAttempts = toRange (Proxy @3) - } hasqlPool = undefined - amqpJobsPublisherChannel = undefined amqpBackendNotificationsChannel = undefined federationDomain = Domain "local" postgresMigration = diff --git a/services/background-worker/test/Test/Wire/Util.hs b/services/background-worker/test/Test/Wire/Util.hs index 5d89532bfec..184d9556299 100644 --- a/services/background-worker/test/Test/Wire/Util.hs +++ b/services/background-worker/test/Test/Wire/Util.hs @@ -22,8 +22,6 @@ module Test.Wire.Util where import Data.Default import Data.Domain (Domain (Domain)) import Data.Misc -import Data.Proxy -import Data.Range import Data.Secret (secretText) import Imports import Network.HTTP.Client hiding (Proxy) @@ -62,14 +60,7 @@ testEnv = do rabbitmqVHost = undefined defederationTimeout = responseTimeoutNone backendNotificationsConfig = BackendNotificationsConfig 1000 500000 1000 - backgroundJobsConfig = - BackgroundJobsConfig - { concurrency = toRange (Proxy @1), - jobTimeout = Duration 100, - maxAttempts = toRange (Proxy @3) - } hasqlPool = undefined - amqpJobsPublisherChannel = undefined amqpBackendNotificationsChannel = undefined federationDomain = Domain "local" gundeckEndpoint = undefined 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 $ From 33f52a246b84404967bc78e517283f751147679a Mon Sep 17 00:00:00 2001 From: Gautier DI FOLCO Date: Tue, 1 Sep 2026 10:43:07 +0200 Subject: [PATCH 2/4] [WPB-28375] Keep RabbitMQ background-jobs consumer for one transitional release Review feedback on the Arbiter move: a helm upgrade replaces brig and background-worker atomically, so operators cannot drain the RabbitMQ background-jobs queue before upgrading. Restore the pre-PR consumer (wire format, RabbitMQ publisher interpreter, background-worker consumer path, backgroundJobs config plumbing, charts/docs) so leftover jobs are drained instead of dropped; brig keeps publishing to Arbiter only. The consumer is removed in the next release; this release must not be skipped. Changelog spells out the drain-and-verify rollout steps. --- .../2-features/WPB-28375-user-groups-arbiter | 10 +- charts/integration/templates/configmap.yaml | 6 + .../background-worker/configmap.yaml | 4 + charts/wire-server/values.yaml | 9 ++ .../src/developer/reference/config-options.md | 19 ++- hack/helm_vars/wire-server/values.yaml.gotmpl | 4 + integration/test/Test/Events.hs | 7 +- libs/wire-api/src/Wire/API/BackgroundJobs.hs | 100 ++++++++++++ .../unit/Test/Wire/API/Roundtrip/Aeson.hs | 2 + .../Wire/BackgroundJobsPublisher/RabbitMQ.hs | 67 ++++++++ libs/wire-subsystems/wire-subsystems.cabal | 1 + .../background-worker/background-worker.cabal | 4 + .../background-worker.integration.yaml | 6 + services/background-worker/default.nix | 4 + .../src/Wire/BackgroundWorker.hs | 8 +- .../src/Wire/BackgroundWorker/Env.hs | 14 +- .../Wire/BackgroundWorker/Jobs/Consumer.hs | 147 ++++++++++++++++++ .../Wire/BackgroundWorker/Jobs/Registry.hs | 41 +++++ .../src/Wire/BackgroundWorker/Options.hs | 14 +- .../Wire/BackendNotificationPusherSpec.hs | 14 ++ .../background-worker/test/Test/Wire/Util.hs | 9 ++ 21 files changed, 476 insertions(+), 14 deletions(-) create mode 100644 libs/wire-subsystems/src/Wire/BackgroundJobsPublisher/RabbitMQ.hs create mode 100644 services/background-worker/src/Wire/BackgroundWorker/Jobs/Consumer.hs create mode 100644 services/background-worker/src/Wire/BackgroundWorker/Jobs/Registry.hs diff --git a/changelog.d/2-features/WPB-28375-user-groups-arbiter b/changelog.d/2-features/WPB-28375-user-groups-arbiter index 192e6631f97..c3d3e2ee5a3 100644 --- a/changelog.d/2-features/WPB-28375-user-groups-arbiter +++ b/changelog.d/2-features/WPB-28375-user-groups-arbiter @@ -1,3 +1,9 @@ -Move the sync-user-group family of background jobs from the RabbitMQ `background-jobs` queue to the Arbiter Postgres 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 consumer, publisher channel, and `backgroundJobs` configuration are removed; backend-notification push and dead-user-notification watching remain on RabbitMQ. +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 note for operators: deploy only after the RabbitMQ `background-jobs` queue has drained to zero. Jobs still in that queue when the last old background-worker pod terminates are dropped; sync jobs are idempotent reconciliations re-triggered by the next SCIM write, so bounded loss is acceptable. +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/charts/integration/templates/configmap.yaml b/charts/integration/templates/configmap.yaml index 04e4bd16f11..02a9d4db410 100644 --- a/charts/integration/templates/configmap.yaml +++ b/charts/integration/templates/configmap.yaml @@ -56,6 +56,12 @@ data: backgroundWorker: host: background-worker.{{ .Release.Namespace }}.svc.cluster.local port: 8080 + # Background jobs defaults for integration tests + backgroundJobs: + concurrency: 4 + jobTimeout: 5s + maxAttempts: 3 + stern: host: stern.{{ .Release.Namespace }}.svc.cluster.local port: 8080 diff --git a/charts/wire-server/templates/background-worker/configmap.yaml b/charts/wire-server/templates/background-worker/configmap.yaml index 34a644af6cf..299d0703d3a 100644 --- a/charts/wire-server/templates/background-worker/configmap.yaml +++ b/charts/wire-server/templates/background-worker/configmap.yaml @@ -90,6 +90,10 @@ data: backendNotificationPusher: {{ toYaml .backendNotificationPusher | indent 6 }} + {{- with .backgroundJobs }} + backgroundJobs: +{{ toYaml . | indent 6 }} + {{- end }} jobs: # Arbiter dispatcher poll interval for all jobs. # Lower values reduce discovery latency, but increase DB polling. diff --git a/charts/wire-server/values.yaml b/charts/wire-server/values.yaml index 178a07dd372..c6dda2176a2 100644 --- a/charts/wire-server/values.yaml +++ b/charts/wire-server/values.yaml @@ -1019,6 +1019,15 @@ background-worker: pushBackoffMaxWait: 300000000 # microseconds, so 300s remotesRefreshInterval: 300000000 # microseconds, so 300s + # Background jobs consumer configuration + backgroundJobs: + # Maximum number of in-flight jobs per process + concurrency: 8 + # Per-attempt timeout in seconds + jobTimeout: 60s + # Total attempts, including the first try + maxAttempts: 3 + # Job dispatcher configuration. jobs: # Arbiter dispatcher poll interval for all jobs. diff --git a/docs/src/developer/reference/config-options.md b/docs/src/developer/reference/config-options.md index a41ef3ab30f..acb9e6b778a 100644 --- a/docs/src/developer/reference/config-options.md +++ b/docs/src/developer/reference/config-options.md @@ -2282,6 +2282,12 @@ migrationOptions: # Required migration timeout for a single migration attempt. timeout: 5s +# Background jobs consumer +backgroundJobs: + concurrency: 8 # in-flight jobs per process + jobTimeout: 60s # per attempt + maxAttempts: 3 # total attempts incl. first run + # Jobs jobs: pollInterval: 5s # how often due jobs are discovered @@ -2316,11 +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`. -`jobs` configures the Arbiter-backed PostgreSQL job queues (meetings, -conversations, and user groups), covering immediate and scheduled or recurring -jobs, and controls their dispatcher, worker-pool, visibility, retry, and -reaper behavior. User-group synchronization jobs run on the `user-groups` -queue; jobs for the same user group are serialized via their group key. +`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 @@ -2357,6 +2365,7 @@ 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 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. diff --git a/hack/helm_vars/wire-server/values.yaml.gotmpl b/hack/helm_vars/wire-server/values.yaml.gotmpl index 14832a6f253..673cb6da817 100644 --- a/hack/helm_vars/wire-server/values.yaml.gotmpl +++ b/hack/helm_vars/wire-server/values.yaml.gotmpl @@ -650,6 +650,10 @@ background-worker: port: "5432" user: wire-server dbname: wire-server + backgroundJobs: + concurrency: 8 + jobTimeout: 60s + maxAttempts: 3 jobs: pollInterval: 1s # Poll every second so due jobs are discovered promptly in tests workerThreads: 1 diff --git a/integration/test/Test/Events.hs b/integration/test/Test/Events.hs index b060cbcf067..a172cf2ff0f 100644 --- a/integration/test/Test/Events.hs +++ b/integration/test/Test/Events.hs @@ -176,11 +176,12 @@ testTemporaryQueuesAreDeletedAfterUse = do aliceClientQueue = Queue {name = fromString aliceClientQueueName, vhost = fromString beResource.berVHost} deadNotifsQueue = Queue {name = fromString "dead-user-notifications", vhost = fromString beResource.berVHost} cellsEventsQueue = Queue {name = fromString "cells_events", vhost = fromString beResource.berVHost} + backgroundJobsQueue = Queue {name = fromString "background-jobs", vhost = fromString beResource.berVHost} -- Wait for queue for the new client to be created eventually $ do queuesBeforeWS <- rabbitmqAdmin.listQueuesByVHost (fromString beResource.berVHost) (fromString "") True 100 1 - queuesBeforeWS.items `shouldMatchSet` [deadNotifsQueue, cellsEventsQueue, aliceClientQueue] + queuesBeforeWS.items `shouldMatchSet` [deadNotifsQueue, cellsEventsQueue, aliceClientQueue, backgroundJobsQueue] runCodensity (createEventsWebSocket alice Nothing) $ \ws -> do handle <- randomHandle @@ -188,7 +189,7 @@ testTemporaryQueuesAreDeletedAfterUse = do queuesDuringWS <- rabbitmqAdmin.listQueuesByVHost (fromString beResource.berVHost) (fromString "") True 100 1 addJSONToFailureContext "queuesDuringWS" queuesDuringWS $ do - length queuesDuringWS.items `shouldMatchInt` 4 + length queuesDuringWS.items `shouldMatchInt` 5 -- We cannot use 'assertEvent' here because there is a race between the temp -- queue being created and rabbitmq fanning out the previous events. @@ -202,7 +203,7 @@ testTemporaryQueuesAreDeletedAfterUse = do eventually $ do queuesAfterWS <- rabbitmqAdmin.listQueuesByVHost (fromString beResource.berVHost) (fromString "") True 100 1 - queuesAfterWS.items `shouldMatchSet` [deadNotifsQueue, cellsEventsQueue, aliceClientQueue] + queuesAfterWS.items `shouldMatchSet` [deadNotifsQueue, cellsEventsQueue, aliceClientQueue, backgroundJobsQueue] testSendMessageNoReturnToSenderWithConsumableNotificationsProteus :: (HasCallStack) => App () testSendMessageNoReturnToSenderWithConsumableNotificationsProteus = withModifiedBackend (enableConsumableNotifications def) $ \domain -> do diff --git a/libs/wire-api/src/Wire/API/BackgroundJobs.hs b/libs/wire-api/src/Wire/API/BackgroundJobs.hs index 772073e2be0..ac99d5dd33e 100644 --- a/libs/wire-api/src/Wire/API/BackgroundJobs.hs +++ b/libs/wire-api/src/Wire/API/BackgroundJobs.hs @@ -1,4 +1,5 @@ {-# LANGUAGE StrictData #-} +{-# LANGUAGE TemplateHaskell #-} -- This file is part of the Wire Server implementation. -- @@ -19,11 +20,16 @@ module Wire.API.BackgroundJobs where +import Control.Arrow ((&&&)) +import Control.Lens (makePrisms) import Data.Aeson qualified as Aeson import Data.Id +import Data.Map.Strict qualified as Map import Data.OpenApi qualified as S import Data.Schema import Imports +import Network.AMQP qualified as Q +import Network.AMQP.Types qualified as QT import Wire.Arbitrary (Arbitrary (..), GenericUniform (..)) data BackgroundJobPayload @@ -32,6 +38,34 @@ data BackgroundJobPayload deriving stock (Eq, Show, Generic) deriving (Arbitrary) via GenericUniform BackgroundJobPayload +backgroundJobPayloadLabel :: BackgroundJobPayload -> Text +backgroundJobPayloadLabel p = case backgroundJobPayloadTag p of + BackgroundJobSyncUserGroupAndChannelTag -> "sync-user-group-and-channel" + BackgroundJobSyncUserGroupTag -> "sync-user-group" + +data BackgroundJobPayloadTag + = BackgroundJobSyncUserGroupAndChannelTag + | BackgroundJobSyncUserGroupTag + deriving stock (Eq, Ord, Bounded, Enum, Show, Generic) + deriving (Arbitrary) via GenericUniform BackgroundJobPayloadTag + +instance ToSchema BackgroundJobPayloadTag where + schema = + enum @Text $ + mconcat + [ element "sync-user-group-and-channel" BackgroundJobSyncUserGroupAndChannelTag, + element "sync-user-group" BackgroundJobSyncUserGroupTag + ] + +backgroundJobPayloadTag :: BackgroundJobPayload -> BackgroundJobPayloadTag +backgroundJobPayloadTag = + \case + BackgroundJobSyncUserGroupAndChannel {} -> BackgroundJobSyncUserGroupAndChannelTag + BackgroundJobSyncUserGroup {} -> BackgroundJobSyncUserGroupTag + +backgroundJobPayloadTagSchema :: ObjectSchema SwaggerDoc BackgroundJobPayloadTag +backgroundJobPayloadTagSchema = field "type" schema + data SyncUserGroupAndChannel = SyncUserGroupAndChannel { teamId :: TeamId, userGroupId :: UserGroupId, @@ -67,3 +101,69 @@ instance ToSchema SyncUserGroup where <$> (.teamId) .= field "team_id" schema <*> (.userGroupId) .= field "user_group_id" schema <*> (.actor) .= maybe_ (optField "actor" schema) + +makePrisms ''BackgroundJobPayload + +backgroundJobPayloadObjectSchema :: ObjectSchema SwaggerDoc BackgroundJobPayload +backgroundJobPayloadObjectSchema = + snd + <$> (backgroundJobPayloadTag &&& id) + .= bind + (fst .= backgroundJobPayloadTagSchema) + (snd .= dispatch backgroundJobPayloadDataSchema) + where + backgroundJobPayloadDataSchema :: BackgroundJobPayloadTag -> ObjectSchema SwaggerDoc BackgroundJobPayload + backgroundJobPayloadDataSchema = \case + BackgroundJobSyncUserGroupAndChannelTag -> tag _BackgroundJobSyncUserGroupAndChannel (field "payload" schema) + BackgroundJobSyncUserGroupTag -> tag _BackgroundJobSyncUserGroup (field "payload" schema) + +instance ToSchema BackgroundJobPayload where + schema = object backgroundJobPayloadObjectSchema + +deriving via (Schema BackgroundJobPayload) instance Aeson.FromJSON BackgroundJobPayload + +deriving via (Schema BackgroundJobPayload) instance Aeson.ToJSON BackgroundJobPayload + +deriving via (Schema BackgroundJobPayload) instance S.ToSchema BackgroundJobPayload + +-- | Background job envelope. Payload is a free-form JSON object. +data BackgroundJob = BackgroundJob + { jobId :: JobId, + requestId :: RequestId, + payload :: BackgroundJobPayload + } + deriving stock (Eq, Show, Generic) + deriving (Arbitrary) via GenericUniform BackgroundJob + deriving (Aeson.ToJSON, Aeson.FromJSON, S.ToSchema) via Schema BackgroundJob + +instance ToSchema BackgroundJob where + schema = + object $ + BackgroundJob + <$> jobId .= field "id" schema + <*> requestId .= field "requestId" schema + <*> payload .= field "payload" schema + +backgroundJobsRoutingKey :: Text +backgroundJobsRoutingKey = backgroundJobsQueueName + +backgroundJobsQueueName :: Text +backgroundJobsQueueName = "background-jobs" + +ensureBackgroundJobsQueue :: Q.Channel -> IO () +ensureBackgroundJobsQueue chan = do + let headers = + QT.FieldTable + ( Map.fromList + [ ("x-queue-type", QT.FVString "quorum") + ] + ) + q = + Q.newQueue + { Q.queueName = backgroundJobsQueueName, + Q.queueDurable = True, + Q.queueAutoDelete = False, + Q.queueExclusive = False, + Q.queueHeaders = headers + } + void $ Q.declareQueue chan q diff --git a/libs/wire-api/test/unit/Test/Wire/API/Roundtrip/Aeson.hs b/libs/wire-api/test/unit/Test/Wire/API/Roundtrip/Aeson.hs index 0c2a791f0a6..14f6f83c610 100644 --- a/libs/wire-api/test/unit/Test/Wire/API/Roundtrip/Aeson.hs +++ b/libs/wire-api/test/unit/Test/Wire/API/Roundtrip/Aeson.hs @@ -31,6 +31,7 @@ import Test.Tasty.HUnit (assertEqual, assertFailure, testCase) import Test.Tasty.QuickCheck (Arbitrary, counterexample, testProperty, (.&&.), (===)) import Type.Reflection (typeRep) import Wire.API.Asset qualified as Asset +import Wire.API.BackgroundJobs qualified as BackgroundJobs import Wire.API.Call.Config qualified as Call.Config import Wire.API.Connection qualified as Connection import Wire.API.Conversation qualified as Conversation @@ -383,6 +384,7 @@ tests = testRoundTrip @TeamsIntra.TeamStatusUpdate, testRoundTrip @TeamsIntra.TeamData, testRoundTrip @TeamsIntra.TeamName, + testRoundTrip @BackgroundJobs.BackgroundJob, testRoundTrip @User.ManagedByUpdate, testRoundTrip @User.Auth.ReAuthUser, testRoundTrip @User.RichInfoUpdate, diff --git a/libs/wire-subsystems/src/Wire/BackgroundJobsPublisher/RabbitMQ.hs b/libs/wire-subsystems/src/Wire/BackgroundJobsPublisher/RabbitMQ.hs new file mode 100644 index 00000000000..6a6719b4fff --- /dev/null +++ b/libs/wire-subsystems/src/Wire/BackgroundJobsPublisher/RabbitMQ.hs @@ -0,0 +1,67 @@ +-- This file is part of the Wire Server implementation. +-- +-- Copyright (C) 2025 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.RabbitMQ where + +import Data.Aeson qualified as Aeson +import Data.Id (JobId, RequestId (..), idToText, randomId) +import Data.Text.Encoding qualified as T +import Imports +import Network.AMQP qualified as Q +import Polysemy +import Wire.API.BackgroundJobs +import Wire.BackgroundJobsPublisher (BackgroundJobPublisher (..)) + +interpretBackgroundJobPublisherRabbitMQ :: + (Member (Embed IO) r) => + RequestId -> + MVar Q.Channel -> + InterpreterFor BackgroundJobPublisher r +interpretBackgroundJobPublisherRabbitMQ requestId channelMVar = + interpret $ \case + PublishJob jobPayload -> do + jobId <- randomId + channel <- readMVar channelMVar + publishJob requestId channel jobId jobPayload + +publishJob :: + (Member (Embed IO) r) => + RequestId -> + Q.Channel -> + JobId -> + BackgroundJobPayload -> + Sem r () +publishJob requestId channel jobId jobPayload = do + let job = + BackgroundJob + { payload = jobPayload, + jobId = jobId, + requestId = requestId + } + msg = + Q.newMsg + { Q.msgBody = Aeson.encode job, + Q.msgContentType = Just "application/json", + Q.msgID = Just (idToText job.jobId), + Q.msgCorrelationID = Just $ T.decodeUtf8 job.requestId.unRequestId + } + + liftIO $ do + ensureBackgroundJobsQueue channel + -- Passing "" for `exchangeName` publishes to the default exchange in RabbitMQ. + -- The default exchange routes directly to the queue whose name equals the `routingKey`. + void $ Q.publishMsg channel "" backgroundJobsRoutingKey msg diff --git a/libs/wire-subsystems/wire-subsystems.cabal b/libs/wire-subsystems/wire-subsystems.cabal index 07b2d84c4e5..e32d082bc34 100644 --- a/libs/wire-subsystems/wire-subsystems.cabal +++ b/libs/wire-subsystems/wire-subsystems.cabal @@ -237,6 +237,7 @@ library Wire.BackendNotificationQueueAccess.RabbitMq Wire.BackgroundJobsPublisher Wire.BackgroundJobsPublisher.Arbiter + Wire.BackgroundJobsPublisher.RabbitMQ Wire.BackgroundJobsRunner Wire.BackgroundJobsRunner.Interpreter Wire.BlockListStore diff --git a/services/background-worker/background-worker.cabal b/services/background-worker/background-worker.cabal index ccf77722432..a43796a4ca7 100644 --- a/services/background-worker/background-worker.cabal +++ b/services/background-worker/background-worker.cabal @@ -17,6 +17,8 @@ library Wire.BackgroundWorker Wire.BackgroundWorker.Env Wire.BackgroundWorker.Health + Wire.BackgroundWorker.Jobs.Consumer + Wire.BackgroundWorker.Jobs.Registry Wire.BackgroundWorker.Options Wire.BackgroundWorker.Util Wire.BackgroundWorker.Workers @@ -44,8 +46,10 @@ library , cassandra-util , containers , cron + , data-timeout , exceptions , extended + , extra , galley-types , hasql-resource-pool , HsOpenSSL diff --git a/services/background-worker/background-worker.integration.yaml b/services/background-worker/background-worker.integration.yaml index 9b89ac94b12..b0bd0d172e4 100644 --- a/services/background-worker/background-worker.integration.yaml +++ b/services/background-worker/background-worker.integration.yaml @@ -60,6 +60,12 @@ migrateTeamFeatures: false migrateDomainRegistration: false migrateUsers: false +# Background jobs consumer configuration for integration +backgroundJobs: + concurrency: 4 + jobTimeout: 5s + maxAttempts: 3 + # Job dispatcher configuration for integration jobs: pollInterval: 1s # Poll every second so due jobs are discovered promptly in tests diff --git a/services/background-worker/default.nix b/services/background-worker/default.nix index 06ff0a24ebb..acf79b6e195 100644 --- a/services/background-worker/default.nix +++ b/services/background-worker/default.nix @@ -15,8 +15,10 @@ , containers , cron , data-default +, data-timeout , exceptions , extended +, extra , federator , galley-types , hasql-resource-pool @@ -75,8 +77,10 @@ mkDerivation { cassandra-util containers cron + data-timeout exceptions extended + extra galley-types hasql-resource-pool HsOpenSSL diff --git a/services/background-worker/src/Wire/BackgroundWorker.hs b/services/background-worker/src/Wire/BackgroundWorker.hs index c2c32df5e02..904d408a994 100644 --- a/services/background-worker/src/Wire/BackgroundWorker.hs +++ b/services/background-worker/src/Wire/BackgroundWorker.hs @@ -29,6 +29,7 @@ import Servant.Server.Generic import UnliftIO (Concurrently (..), runConcurrently) import Util.Options import Wire.BackendNotificationPusher qualified as BackendNotificationPusher +import Wire.BackgroundWorker.Jobs.Consumer qualified as Jobs import Wire.BackgroundWorker.Env import Wire.BackgroundWorker.Health qualified as Health import Wire.BackgroundWorker.Options @@ -88,10 +89,14 @@ run opts galleyOpts = do 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 $ - (,,,,,,,,) + (,,,,,,,) <$> Concurrently cleanupDeadUserNotifWatcher <*> Concurrently cleanupBackendNotifPusher <*> Concurrently cleanupConvMigration @@ -100,6 +105,7 @@ run opts galleyOpts = do <*> Concurrently cleanupDomainRegistrationMigration <*> Concurrently cleanupUsersMigration <*> Concurrently cleanupJobRunner + <*> Concurrently cleanupJobs let server = defaultServer (T.unpack opts.backgroundWorker.host) opts.backgroundWorker.port env.logger let settings = newSettings server diff --git a/services/background-worker/src/Wire/BackgroundWorker/Env.hs b/services/background-worker/src/Wire/BackgroundWorker/Env.hs index 2fe576c57ed..ed784a33db9 100644 --- a/services/background-worker/src/Wire/BackgroundWorker/Env.hs +++ b/services/background-worker/src/Wire/BackgroundWorker/Env.hs @@ -64,12 +64,14 @@ type IsWorking = Bool data Worker = BackendNotificationPusher | DeadUserNotificationWatcher + | BackgroundJobConsumer deriving (Eq, Ord) workerName :: Worker -> Text workerName = \case BackendNotificationPusher -> "backend-notification-pusher" DeadUserNotificationWatcher -> "dead-user-notification-watcher" + BackgroundJobConsumer -> "background-job-consumer" data Env = Env { http2Manager :: Http2Manager, @@ -82,6 +84,7 @@ data Env = Env backendNotificationMetrics :: BackendNotificationMetrics, meetingsCleanupMetrics :: MeetingsCleanupMetrics, backendNotificationsConfig :: BackendNotificationsConfig, + backgroundJobsConfig :: BackgroundJobsConfig, workerRunningGauge :: Vector Text Gauge, statuses :: IORef (Map Worker IsWorking), cassandra :: ClientState, @@ -90,7 +93,8 @@ data Env = Env hasqlPool :: Hasql.Pool, -- May contain the PostgreSQL password. Do not unwrap outside the Arbiter boundary. arbiterConnStr :: SecretText, - -- Dedicated AMQP channel per concern + -- Dedicated AMQP channels per concern + amqpJobsPublisherChannel :: MVar Q.Channel, amqpBackendNotificationsChannel :: MVar Q.Channel, federationDomain :: Domain, postgresMigration :: PostgresMigrationOpts, @@ -166,11 +170,13 @@ mkEnv opts galleyOpts = do statuses <- newIORef $ Map.fromList - [ (BackendNotificationPusher, False) + [ (BackendNotificationPusher, False), + (BackgroundJobConsumer, False) ] backendNotificationMetrics <- mkBackendNotificationMetrics meetingsCleanupMetrics <- mkMeetingsCleanupMetrics let backendNotificationsConfig = opts.backendNotificationPusher + backgroundJobsConfig = opts.backgroundJobs federationDomain = galleyOpts._settings._federationDomain postgresMigration = opts.postgresMigration brigEndpoint = opts.brig @@ -189,6 +195,10 @@ mkEnv opts galleyOpts = do workerRunningGauge <- mkWorkerRunningGauge hasqlPool <- initPostgresPool opts.postgresqlPool galleyOpts._postgresql galleyOpts._postgresqlPassword arbiterConnStr <- mkArbiterConnectionString galleyOpts._postgresql galleyOpts._postgresqlPassword + Log.info logger $ Log.msg @Text "Opening RabbitMQ channel: background-worker-jobs-publisher..." + amqpJobsPublisherChannel <- + mkRabbitMqChannelMVar logger (Just "background-worker-jobs-publisher") $ + either id demoteOpts opts.rabbitmq.unRabbitMqOpts Log.info logger $ Log.msg @Text "Opening RabbitMQ channel: background-worker-backend-notifications..." amqpBackendNotificationsChannel <- mkRabbitMqChannelMVar logger (Just "background-worker-backend-notifications") $ diff --git a/services/background-worker/src/Wire/BackgroundWorker/Jobs/Consumer.hs b/services/background-worker/src/Wire/BackgroundWorker/Jobs/Consumer.hs new file mode 100644 index 00000000000..cb0d8a4b9c6 --- /dev/null +++ b/services/background-worker/src/Wire/BackgroundWorker/Jobs/Consumer.hs @@ -0,0 +1,147 @@ +{-# LANGUAGE RecordWildCards #-} + +-- This file is part of the Wire Server implementation. +-- +-- Copyright (C) 2025 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.BackgroundWorker.Jobs.Consumer (startWorker, BackgroundJobMetrics (..)) where + +import Control.Concurrent.Timeout qualified as Timeout +import Control.Retry +import Data.Aeson qualified as Aeson +import Data.Misc hiding (duration) +import Data.Range (Range (fromRange)) +import Data.Timeout +import Imports +import Network.AMQP qualified as Q +import Network.AMQP.Extended +import Network.AMQP.Lifted qualified as QL +import Prometheus +import System.Logger.Class qualified as Log +import System.Time.Extra (duration) +import UnliftIO +import Wire.API.BackgroundJobs +import Wire.BackgroundWorker.Env +import Wire.BackgroundWorker.Jobs.Registry +import Wire.BackgroundWorker.Options +import Wire.BackgroundWorker.Util (CleanupAction) + +data BackgroundJobMetrics = BackgroundJobMetrics + { workersBusy :: Gauge, + concurrencyConfigured :: Gauge, + jobsReceived :: Vector Text Counter, + jobsStarted :: Vector Text Counter, + jobsSucceeded :: Vector Text Counter, + jobsFailed :: Vector Text Counter, + jobsInvalid :: Vector Text Counter, + jobsRedelivered :: Vector Text Counter, + jobDuration :: Vector Text Histogram + } + +mkMetrics :: IO BackgroundJobMetrics +mkMetrics = do + workersBusy <- register (gauge $ Info {metricName = "wire_background_jobs_workers_busy", metricHelp = "In-flight background jobs"}) + concurrencyConfigured <- register (gauge $ Info {metricName = "wire_background_jobs_concurrency_configured", metricHelp = "Configured concurrency for this process"}) + jobsReceived <- register (vector "job_type" $ counter $ Info "wire_background_jobs_received_total" "Jobs received") + jobsStarted <- register (vector "job_type" $ counter $ Info "wire_background_jobs_started_total" "Jobs started") + jobsSucceeded <- register (vector "job_type" $ counter $ Info "wire_background_jobs_succeeded_total" "Jobs succeeded") + jobsFailed <- register (vector "job_type" $ counter $ Info "wire_background_jobs_failed_total" "Jobs failed") + jobsInvalid <- register (vector "job_type" $ counter $ Info "wire_background_jobs_invalid_total" "Invalid jobs received") + jobsRedelivered <- register (vector "job_type" $ counter $ Info "wire_background_jobs_redelivered_total" "Jobs marked redelivered by broker") + jobDuration <- register (vector "job_type" $ histogram (Info "wire_background_jobs_duration_seconds" "Job duration seconds") defaultBuckets) + pure BackgroundJobMetrics {..} + +startWorker :: AmqpEndpoint -> AppT IO CleanupAction +startWorker rabbitmqOpts = do + env <- ask + let cfg = env.backgroundJobsConfig + metrics <- liftIO mkMetrics + markAsNotWorking BackgroundJobConsumer + void . async . liftIO $ + openConnectionWithRetries env.logger rabbitmqOpts (Just "background-job-consumer") $ + RabbitMqHooks + { onNewChannel = \chan -> do + -- declare queue and set prefetch to concurrency + ensureBackgroundJobsQueue chan + Q.qos chan 0 (fromIntegral $ fromRange cfg.concurrency) False + -- set gauges + setGauge metrics.concurrencyConfigured (fromIntegral $ fromRange cfg.concurrency) + -- start consuming with manual ack and keep the channel alive + void $ QL.consumeMsgs chan backgroundJobsQueueName Q.Ack (void . runAppT env . handleDelivery metrics cfg) + runAppT env $ markAsWorking BackgroundJobConsumer + forever $ threadDelay maxBound, + onChannelException = \e -> do + runAppT env $ markAsNotWorking BackgroundJobConsumer + let connClosed = + case (fromException e :: Maybe Q.AMQPException) of + Just (Q.ConnectionClosedException _ _) -> True + _ -> False + unless (Q.isNormalChannelClose e || connClosed) $ + runAppT env $ + Log.err $ + Log.msg (Log.val "Unexpected RabbitMQ channel exception in background job consumer") + . Log.field "exception" (displayException e), + onConnectionClose = + runAppT env $ do + markAsNotWorking BackgroundJobConsumer + Log.info $ Log.msg (Log.val "RabbitMQ connection closed for background job consumer") + } + pure $ runAppT env $ cleanup + where + cleanup :: AppT IO () + cleanup = do + -- nothing to close explicitly; the AMQP helper closes channel/connection on shutdown + Log.info $ Log.msg (Log.val "Background job consumer cleanup") + markAsNotWorking BackgroundJobConsumer + +handleDelivery :: BackgroundJobMetrics -> BackgroundJobsConfig -> (Q.Message, Q.Envelope) -> AppT IO () +handleDelivery metrics cfg (msg, env) = do + case Aeson.eitherDecode @BackgroundJob (Q.msgBody msg) of + Left err -> do + withLabel metrics.jobsInvalid "invalid" incCounter + Log.err $ Log.msg (Log.val "Invalid background job JSON") . Log.field "error" err + Timeout.threadDelay (200 # MilliSecond) -- avoid tight redelivery loop + liftIO $ Q.rejectEnv env True + Right job -> do + let lbl = backgroundJobPayloadLabel job.payload + when (Q.envRedelivered env) $ withLabel metrics.jobsRedelivered lbl incCounter + withLabel metrics.jobsReceived lbl incCounter + UnliftIO.bracket_ (incGauge metrics.workersBusy) (decGauge metrics.workersBusy) $ do + outcome <- runAttempts lbl job + case outcome of + Right () -> do + withLabel metrics.jobsSucceeded lbl incCounter + liftIO $ Q.ackEnv env + Left e -> do + withLabel metrics.jobsFailed lbl incCounter + Log.err $ Log.msg (Log.val "Background job failed after retries") . Log.field "error" e + liftIO $ Q.rejectEnv env False + where + runAttempts :: Text -> BackgroundJob -> AppT IO (Either Text ()) + runAttempts lbl job = do + let retries = max 0 (fromRange cfg.maxAttempts - 1) + policy = limitRetries retries <> fullJitterBackoff 100000 -- 100ms base + retrying policy shouldRetry $ \_rs -> do + withLabel metrics.jobsStarted lbl incCounter + (dur, r) <- + duration $ + fromMaybe (Left "job timeout") + <$> timeout (durationToMicros cfg.jobTimeout) (dispatchJob job) + withLabel metrics.jobDuration lbl (`observe` dur) + pure r + where + shouldRetry _ (Right _) = pure False + shouldRetry _ (Left _) = pure True diff --git a/services/background-worker/src/Wire/BackgroundWorker/Jobs/Registry.hs b/services/background-worker/src/Wire/BackgroundWorker/Jobs/Registry.hs new file mode 100644 index 00000000000..62384214cb9 --- /dev/null +++ b/services/background-worker/src/Wire/BackgroundWorker/Jobs/Registry.hs @@ -0,0 +1,41 @@ +-- This file is part of the Wire Server implementation. +-- +-- Copyright (C) 2025 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.BackgroundWorker.Jobs.Registry + ( dispatchJob, + ) +where + +import Imports +import Wire.API.BackgroundJobs (BackgroundJob (..)) +import Wire.BackgroundJobsPublisher.RabbitMQ (interpretBackgroundJobPublisherRabbitMQ) +import Wire.BackgroundJobsRunner (runJob) +import Wire.BackgroundJobsRunner.Interpreter (interpretBackgroundJobRunner) +import Wire.BackgroundWorker.Env (AppT, Env (..)) +import Wire.Effects +import Wire.ExternalAccess.External + +dispatchJob :: BackgroundJob -> AppT IO (Either Text ()) +dispatchJob job = do + env <- ask @Env + let disableTlsV1 = True + extEnv <- liftIO (initExtEnv disableTlsV1) + liftIO + $ runBackgroundWorkerEffects env extEnv job.requestId (Just job.jobId) + . interpretBackgroundJobPublisherRabbitMQ job.requestId env.amqpJobsPublisherChannel + . interpretBackgroundJobRunner + $ runJob job.payload diff --git a/services/background-worker/src/Wire/BackgroundWorker/Options.hs b/services/background-worker/src/Wire/BackgroundWorker/Options.hs index 0e9891507cf..035460cfc32 100644 --- a/services/background-worker/src/Wire/BackgroundWorker/Options.hs +++ b/services/background-worker/src/Wire/BackgroundWorker/Options.hs @@ -57,7 +57,8 @@ data Opts = Opts migrateDomainRegistration :: !Bool, migrateUsers :: !Bool, jobs :: JobConfig, - meetingsCleanup :: MeetingsCleanupConfig + meetingsCleanup :: MeetingsCleanupConfig, + backgroundJobs :: BackgroundJobsConfig } deriving (Show, Generic) deriving (FromJSON) via Generically Opts @@ -89,6 +90,17 @@ instance FromJSON RabbitMqOpts where <|> (Left <$> parseJSON v) ) +data BackgroundJobsConfig = BackgroundJobsConfig + { -- | Maximum parallel jobs processed by this process + concurrency :: Range 1 1000 Int, + -- | Per-attempt timeout (seconds) + jobTimeout :: Duration, + -- | Total attempts including first run + maxAttempts :: Range 1 1000 Int + } + deriving (Show, Generic) + deriving (FromJSON) via Generically BackgroundJobsConfig + data JobConfig = JobConfig { -- | Arbiter dispatcher poll interval for jobs. -- Lower values reduce discovery latency for due jobs. diff --git a/services/background-worker/test/Test/Wire/BackendNotificationPusherSpec.hs b/services/background-worker/test/Test/Wire/BackendNotificationPusherSpec.hs index 66befeb0a03..7222120d93a 100644 --- a/services/background-worker/test/Test/Wire/BackendNotificationPusherSpec.hs +++ b/services/background-worker/test/Test/Wire/BackendNotificationPusherSpec.hs @@ -355,7 +355,14 @@ spec = do rabbitmqVHost = "test-vhost" defederationTimeout = responseTimeoutNone backendNotificationsConfig = BackendNotificationsConfig 1000 500000 1000 + backgroundJobsConfig = + BackgroundJobsConfig + { concurrency = toRange (Proxy @1), + jobTimeout = Duration 100, + maxAttempts = toRange (Proxy @3) + } hasqlPool = undefined + amqpJobsPublisherChannel = undefined amqpBackendNotificationsChannel = undefined federationDomain = Domain "local" postgresMigration = @@ -412,7 +419,14 @@ spec = do rabbitmqVHost = "test-vhost" defederationTimeout = responseTimeoutNone backendNotificationsConfig = BackendNotificationsConfig 1000 500000 1000 + backgroundJobsConfig = + BackgroundJobsConfig + { concurrency = toRange (Proxy @1), + jobTimeout = Duration 100, + maxAttempts = toRange (Proxy @3) + } hasqlPool = undefined + amqpJobsPublisherChannel = undefined amqpBackendNotificationsChannel = undefined federationDomain = Domain "local" postgresMigration = diff --git a/services/background-worker/test/Test/Wire/Util.hs b/services/background-worker/test/Test/Wire/Util.hs index 184d9556299..5d89532bfec 100644 --- a/services/background-worker/test/Test/Wire/Util.hs +++ b/services/background-worker/test/Test/Wire/Util.hs @@ -22,6 +22,8 @@ module Test.Wire.Util where import Data.Default import Data.Domain (Domain (Domain)) import Data.Misc +import Data.Proxy +import Data.Range import Data.Secret (secretText) import Imports import Network.HTTP.Client hiding (Proxy) @@ -60,7 +62,14 @@ testEnv = do rabbitmqVHost = undefined defederationTimeout = responseTimeoutNone backendNotificationsConfig = BackendNotificationsConfig 1000 500000 1000 + backgroundJobsConfig = + BackgroundJobsConfig + { concurrency = toRange (Proxy @1), + jobTimeout = Duration 100, + maxAttempts = toRange (Proxy @3) + } hasqlPool = undefined + amqpJobsPublisherChannel = undefined amqpBackendNotificationsChannel = undefined federationDomain = Domain "local" gundeckEndpoint = undefined From f19c69e98be7d2ee40e222b681e040354d93c8ea Mon Sep 17 00:00:00 2001 From: Gautier DI FOLCO Date: Thu, 3 Sep 2026 02:07:42 +0200 Subject: [PATCH 3/4] [WPB-28375] Fix import order for ormolu --- services/background-worker/src/Wire/BackgroundWorker.hs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/services/background-worker/src/Wire/BackgroundWorker.hs b/services/background-worker/src/Wire/BackgroundWorker.hs index 904d408a994..318d35209fa 100644 --- a/services/background-worker/src/Wire/BackgroundWorker.hs +++ b/services/background-worker/src/Wire/BackgroundWorker.hs @@ -29,9 +29,9 @@ import Servant.Server.Generic import UnliftIO (Concurrently (..), runConcurrently) import Util.Options import Wire.BackendNotificationPusher qualified as BackendNotificationPusher -import Wire.BackgroundWorker.Jobs.Consumer qualified as Jobs import Wire.BackgroundWorker.Env import Wire.BackgroundWorker.Health qualified as Health +import Wire.BackgroundWorker.Jobs.Consumer qualified as Jobs import Wire.BackgroundWorker.Options import Wire.BackgroundWorker.Workers as Workers import Wire.DeadUserNotificationWatcher qualified as DeadUserNotificationWatcher @@ -96,7 +96,7 @@ run opts galleyOpts = do let cleanup = void $ runConcurrently $ - (,,,,,,,) + (,,,,,,,,) <$> Concurrently cleanupDeadUserNotifWatcher <*> Concurrently cleanupBackendNotifPusher <*> Concurrently cleanupConvMigration From 1f358dcdd79bd46ef53a11dc29952c9959ddb3fb Mon Sep 17 00:00:00 2001 From: Gautier DI FOLCO Date: Thu, 3 Sep 2026 13:44:45 +0200 Subject: [PATCH 4/4] Hello CI