[WPB 3796] Fix background worker for federation policy allowAll

changelog.d/3-bug-fixes/WPB-3796

@@ -0,0 +1 @@
+[background-worker] Get federation remote domains from rabbitMQ, not from brig (this fixes a bug in federation policy `allowAll`)

services/background-worker/background-worker.cabal

@@ -31,7 +31,6 @@ library
       aeson
     , amqp
     , async
-    , base
     , bilge
     , bytestring-conversion
     , containers
@@ -56,7 +55,6 @@ library
     , types-common
     , unliftio
     , wai-utilities
-    , wire-api
     , wire-api-federation
 
   default-extensions:

services/background-worker/default.nix

@@ -56,7 +56,6 @@ mkDerivation {
     aeson
     amqp
     async
-    base
     bilge
     bytestring-conversion
     containers
@@ -81,7 +80,6 @@ mkDerivation {
     types-common
     unliftio
     wai-utilities
-    wire-api
     wire-api-federation
   ];
   executableHaskellDepends = [ HsOpenSSL imports types-common ];

services/background-worker/src/Wire/BackendNotificationPusher.hs

@@ -21,7 +21,6 @@ import System.Logger.Class qualified as Log
 import UnliftIO
 import Wire.API.Federation.BackendNotifications
 import Wire.API.Federation.Client
-import Wire.API.Routes.FederationDomainConfig
 import Wire.BackgroundWorker.Env
 import Wire.BackgroundWorker.Options
 import Wire.BackgroundWorker.Util
@@ -123,7 +122,7 @@ startPusher consumersRef chan = do
         throwM e
 
   -- If this thread is cancelled, catch the exception, kill the consumers, and carry on.
-  -- FUTUREWORK?:
+  --
   -- If this throws an exception on the Chan / in the forever loop, the exception will
   -- bubble all the way up and kill the pod. Kubernetes should restart the pod automatically.
   flip
@@ -132,29 +131,15 @@ startPusher consumersRef chan = do
       Handler $ cleanup @SomeAsyncException
     ]
     $ do
-      -- Get an initial set of domains from the sync thread
-      -- The Chan that we will be waiting on isn't initialised with a
-      -- value until the domain update loop runs the callback for the
-      -- first time.
-      initRemotes <- liftIO $ readIORef env.remoteDomains
-      -- Get an initial set of consumers for the domains pulled from the IORef
-      -- so that we aren't just sitting around not doing anything for a bit at
-      -- the start.
-      ensureConsumers consumersRef chan $ domain <$> initRemotes.remotes
-      -- Wait for updates to the domains, this is where the bulk of the action
-      -- is going to take place
+      consumers <- newIORef mempty
       forever $ do
-        -- Wait for a new set of domains. This is a blocking action
-        -- so we will only move past here when we get a new set of domains.
-        -- It is a bit nicer than having another timeout value, as Brig is
-        -- already providing one in the domain update message.
-        chanRemotes <- liftIO $ readChan env.remoteDomainsChan
-        -- Make new consumers for the new domains, clean up old ones from the consumer map.
-        ensureConsumers consumersRef chan $ domain <$> chanRemotes.remotes
+        remoteDomains <- getRemoteDomains
+        ensureConsumers consumers chan remoteDomains
+        threadDelay (round $ 1_000_000 * fromMaybe 60 env.backendNotificationsConfig.remotesRefreshInterval)
 
 ensureConsumers :: IORef (Map Domain (Q.ConsumerTag, MVar ())) -> Q.Channel -> [Domain] -> AppT IO ()
 ensureConsumers consumers chan domains = do
-  keys' <- Set.fromList . Map.keys <$> readIORef consumers
+  keys' <- Map.keysSet <$> readIORef consumers
   let domains' = Set.fromList domains
       droppedDomains = Set.difference keys' domains'
   -- Loop over all of the new domains. We can check for existing consumers and add new ones.

services/background-worker/src/Wire/BackgroundWorker.hs

@@ -2,7 +2,6 @@
 
 module Wire.BackgroundWorker where
 
-import Control.Concurrent.Async (cancel)
 import Data.Domain
 import Data.Map.Strict qualified as Map
 import Data.Metrics.Servant qualified as Metrics
@@ -22,14 +21,13 @@ import Wire.Defederation as Defederation
 -- FUTUREWORK: Start an http service with status and metrics endpoints
 run :: Opts -> IO ()
 run opts = do
-  (env, syncThread) <- mkEnv opts
+  env <- mkEnv opts
   (defedChanRef, defedConsumerRef) <- runAppT env $ Defederation.startWorker opts.rabbitmq
   (notifChanRef, notifConsumersRef) <- runAppT env $ BackendNotificationPusher.startWorker opts.rabbitmq
   let -- cleanup will run in a new thread when the signal is caught, so we need to use IORefs and
       -- specific exception types to message threads to clean up
       l = logger env
       cleanup = do
-        cancel syncThread
         -- Cancel the consumers and wait for them to finish their processing step.
         -- Defederation thread
         Log.info (logger env) $ Log.msg (Log.val "Cancelling the defederation thread")

services/background-worker/src/Wire/BackgroundWorker/Env.hs

@@ -3,8 +3,6 @@
 
 module Wire.BackgroundWorker.Env where
 
-import Control.Concurrent.Async
-import Control.Concurrent.Chan
 import Control.Monad.Base
 import Control.Monad.Catch
 import Control.Monad.Trans.Control
@@ -24,8 +22,6 @@ import System.Logger qualified as Log
 import System.Logger.Class (Logger, MonadLogger (..))
 import System.Logger.Extended qualified as Log
 import Util.Options
-import Wire.API.FederationUpdate
-import Wire.API.Routes.FederationDomainConfig
 import Wire.BackgroundWorker.Options
 
 type IsWorking = Bool
@@ -45,10 +41,7 @@ data Env = Env
     federatorInternal :: Endpoint,
     httpManager :: Manager,
     galley :: Endpoint,
-    brig :: Endpoint,
     defederationTimeout :: ResponseTimeout,
-    remoteDomains :: IORef FederationDomainConfigs,
-    remoteDomainsChan :: Chan FederationDomainConfigs,
     backendNotificationMetrics :: BackendNotificationMetrics,
     -- This is needed so that the defederation worker can push
     -- connection-removed notifications into the notifications channels.
@@ -71,27 +64,19 @@ mkBackendNotificationMetrics =
     <*> register (vector "targetDomain" $ counter $ Prometheus.Info "wire_backend_notifications_errors" "Number of errors that occurred while pushing notifications")
     <*> register (vector "targetDomain" $ gauge $ Prometheus.Info "wire_backend_notifications_stuck_queues" "Set to 1 when pushing notifications is stuck")
 
-mkEnv :: Opts -> IO (Env, Async ())
+mkEnv :: Opts -> IO Env
 mkEnv opts = do
   http2Manager <- initHttp2Manager
   logger <- Log.mkLogger opts.logLevel Nothing opts.logFormat
   httpManager <- newManager defaultManagerSettings
-  remoteDomainsChan <- newChan
   let federatorInternal = opts.federatorInternal
       galley = opts.galley
       defederationTimeout =
         maybe
           responseTimeoutNone
           (\t -> responseTimeoutMicro $ 1000000 * t) -- seconds to microseconds
           opts.defederationTimeout
-      brig = opts.brig
       rabbitmqVHost = opts.rabbitmq.vHost
-      callback =
-        SyncFedDomainConfigsCallback
-          { fromFedUpdateCallback = \_old new -> do
-              writeChan remoteDomainsChan new
-          }
-  (remoteDomains, syncThread) <- syncFedDomainConfigs brig logger callback
   rabbitmqAdminClient <- mkRabbitMqAdminClientEnv opts.rabbitmq
   statuses <-
     newIORef $
@@ -103,7 +88,7 @@ mkEnv opts = do
   backendNotificationMetrics <- mkBackendNotificationMetrics
   notificationChannel <- mkRabbitMqChannelMVar logger $ demoteOpts opts.rabbitmq
   let backendNotificationsConfig = opts.backendNotificationPusher
-  pure (Env {..}, syncThread)
+  pure Env {..}
 
 initHttp2Manager :: IO Http2Manager
 initHttp2Manager = do

services/background-worker/src/Wire/BackgroundWorker/Options.hs

@@ -13,7 +13,6 @@ data Opts = Opts
     federatorInternal :: !Endpoint,
     rabbitmq :: !RabbitMqAdminOpts,
     galley :: !Endpoint,
-    brig :: !Endpoint,
     -- | Seconds, Nothing for no timeout
     defederationTimeout :: Maybe Int,
     backendNotificationPusher :: BackendNotificationsConfig
@@ -31,7 +30,9 @@ data BackendNotificationsConfig = BackendNotificationsConfig
     -- | Upper limit on amount of time (in microseconds) to wait before retrying
     -- any notification. This exists to ensure that exponential back-off doesn't
     -- cause wait times to be very big.
-    pushBackoffMaxWait :: Int
+    pushBackoffMaxWait :: Int,
+    -- | Number of seconds between two calls to `getRemoteDomains`.
+    remotesRefreshInterval :: Maybe Double
   }
   deriving (Show, Generic)
 

services/background-worker/src/Wire/Defederation.hs

@@ -7,21 +7,16 @@ import Control.Monad.Catch
 import Control.Retry
 import Data.Aeson qualified as A
 import Data.ByteString.Conversion
-import Data.Domain
-import Data.Text (unpack)
 import Data.Text.Encoding
 import Imports
 import Network.AMQP qualified as Q
 import Network.AMQP.Extended
 import Network.AMQP.Lifted qualified as QL
 import Network.HTTP.Client
 import Network.HTTP.Types
-import Servant.Client (BaseUrl (..), ClientEnv, Scheme (Http), mkClientEnv)
 import System.Logger.Class qualified as Log
-import Util.Options
 import Util.Options qualified as O
 import Wire.API.Federation.BackendNotifications
-import Wire.API.Routes.FederationDomainConfig qualified as Fed
 import Wire.BackgroundWorker.Env
 import Wire.BackgroundWorker.Util
 
@@ -52,18 +47,6 @@ deleteFederationDomainInner' go (msg, envelope) = do
         Log.msg (Log.val "Failed to delete federation domain")
           . Log.field "error" err
 
-mkBrigEnv :: AppT IO ClientEnv
-mkBrigEnv = do
-  Endpoint brigHost brigPort <- asks brig
-  mkClientEnv
-    <$> asks httpManager
-    <*> pure (BaseUrl Http (unpack brigHost) (fromIntegral brigPort) "")
-
-getRemoteDomains :: AppT IO [Domain]
-getRemoteDomains = do
-  ref <- asks remoteDomains
-  fmap Fed.domain . Fed.remotes <$> readIORef ref
-
 callGalleyDelete ::
   ( MonadReader Env m,
     MonadMask m,

services/background-worker/test/Test/Wire/BackendNotificationPusherSpec.hs

@@ -4,7 +4,6 @@
 
 module Test.Wire.BackendNotificationPusherSpec where
 
-import Control.Concurrent.Chan
 import Control.Exception
 import Control.Monad.Trans.Except
 import Data.Aeson qualified as Aeson
@@ -42,7 +41,6 @@ import Wire.API.Federation.API.Brig
 import Wire.API.Federation.API.Common
 import Wire.API.Federation.BackendNotifications
 import Wire.API.RawJson
-import Wire.API.Routes.FederationDomainConfig
 import Wire.BackendNotificationPusher
 import Wire.BackgroundWorker.Env
 import Wire.BackgroundWorker.Options
@@ -220,8 +218,6 @@ spec = do
           ]
       logger <- Logger.new Logger.defSettings
       httpManager <- newManager defaultManagerSettings
-      remoteDomains <- newIORef defFederationDomainConfigs
-      remoteDomainsChan <- newChan
       notificationChannel <- newEmptyMVar
       let federatorInternal = Endpoint "localhost" 8097
           http2Manager = undefined
@@ -231,8 +227,7 @@ spec = do
           rabbitmqVHost = "test-vhost"
           defederationTimeout = responseTimeoutNone
           galley = Endpoint "localhost" 8085
-          brig = Endpoint "localhost" 8082
-          backendNotificationsConfig = BackendNotificationsConfig 1000 500000
+          backendNotificationsConfig = BackendNotificationsConfig 1000 500000 (Just 0.5)
 
       backendNotificationMetrics <- mkBackendNotificationMetrics
       domains <- runAppT Env {..} getRemoteDomains
@@ -243,8 +238,6 @@ spec = do
       mockAdmin <- newMockRabbitMqAdmin True ["backend-notifications.foo.example"]
       logger <- Logger.new Logger.defSettings
       httpManager <- newManager defaultManagerSettings
-      remoteDomains <- newIORef defFederationDomainConfigs
-      remoteDomainsChan <- newChan
       notificationChannel <- newEmptyMVar
       let federatorInternal = Endpoint "localhost" 8097
           http2Manager = undefined
@@ -254,8 +247,7 @@ spec = do
           rabbitmqVHost = "test-vhost"
           defederationTimeout = responseTimeoutNone
           galley = Endpoint "localhost" 8085
-          brig = Endpoint "localhost" 8082
-          backendNotificationsConfig = BackendNotificationsConfig 1000 500000
+          backendNotificationsConfig = BackendNotificationsConfig 1000 500000 (Just 0.5)
       backendNotificationMetrics <- mkBackendNotificationMetrics
       domainsThread <- async $ runAppT Env {..} getRemoteDomains
 

services/background-worker/test/Test/Wire/Util.hs

@@ -2,12 +2,10 @@
 
 module Test.Wire.Util where
 
-import Control.Concurrent.Chan
 import Imports
 import Network.HTTP.Client
 import System.Logger.Class qualified as Logger
 import Util.Options (Endpoint (..))
-import Wire.API.Routes.FederationDomainConfig
 import Wire.BackgroundWorker.Env hiding (federatorInternal, galley)
 import Wire.BackgroundWorker.Env qualified as E
 import Wire.BackgroundWorker.Options
@@ -20,17 +18,14 @@ testEnv = do
   statuses <- newIORef mempty
   backendNotificationMetrics <- mkBackendNotificationMetrics
   httpManager <- newManager defaultManagerSettings
-  remoteDomains <- newIORef defFederationDomainConfigs
-  remoteDomainsChan <- newChan
   notificationChannel <- newEmptyMVar
   let federatorInternal = Endpoint "localhost" 0
       rabbitmqAdminClient = undefined
       rabbitmqVHost = undefined
       metrics = undefined
       galley = Endpoint "localhost" 8085
-      brig = Endpoint "localhost" 8082
       defederationTimeout = responseTimeoutNone
-      backendNotificationsConfig = BackendNotificationsConfig 1000 500000
+      backendNotificationsConfig = BackendNotificationsConfig 1000 500000 (Just 0.5)
   pure Env {..}
 
 runTestAppT :: AppT IO a -> Int -> IO a