diff --git a/changelog.d/6-federation/WPB-28421 b/changelog.d/6-federation/WPB-28421 new file mode 100644 index 00000000000..6425a4060cd --- /dev/null +++ b/changelog.d/6-federation/WPB-28421 @@ -0,0 +1 @@ +Add an opt-in policy for dropping queued federation notifications when the target backend supports no compatible API version. diff --git a/libs/wire-api-federation/src/Wire/API/Federation/BackendNotifications.hs b/libs/wire-api-federation/src/Wire/API/Federation/BackendNotifications.hs index ac1e0e03cd9..9e80eafe1ff 100644 --- a/libs/wire-api-federation/src/Wire/API/Federation/BackendNotifications.hs +++ b/libs/wire-api-federation/src/Wire/API/Federation/BackendNotifications.hs @@ -36,6 +36,7 @@ import Network.AMQP qualified as Q import Network.AMQP.Types qualified as Q import Servant import Servant.Client.Core +import Test.QuickCheck (Arbitrary (arbitrary), elements) import Wire.API.Federation.API.Common import Wire.API.Federation.Client import Wire.API.Federation.Component @@ -75,6 +76,29 @@ instance ToSchema BackendNotification where <*> bodyVersions .= maybe_ (optField "bodyVersions" schema) <*> (.requestId) .= maybe_ (optField "requestId" schema) +data UnsupportedVersionPolicy + = KeepQueued + | DropIfUnsupported + deriving stock (Eq, Show) + deriving (A.ToJSON, A.FromJSON) via (Schema UnsupportedVersionPolicy) + +instance Arbitrary UnsupportedVersionPolicy where + arbitrary = elements [KeepQueued, DropIfUnsupported] + +instance ToSchema UnsupportedVersionPolicy where + schema = + enum @Text $ + mconcat + [ element "keep_queued" KeepQueued, + element "drop_if_unsupported" DropIfUnsupported + ] + +-- Keeping the notification queued is the safe choice if representations +-- configured with different policies are accidentally combined. +instance Semigroup UnsupportedVersionPolicy where + DropIfUnsupported <> DropIfUnsupported = DropIfUnsupported + _ <> _ = KeepQueued + -- | Convert a federation endpoint to a backend notification to be enqueued to a -- RabbitMQ queue. fedNotifToBackendNotif :: @@ -104,17 +128,29 @@ fedNotifToBackendNotif rid ownDomain payload = requestId = Just rid } -newtype PayloadBundle (c :: Component) = PayloadBundle - { notifications :: NE.NonEmpty BackendNotification +data PayloadBundle (c :: Component) = PayloadBundle + { notifications :: NE.NonEmpty BackendNotification, + unsupportedVersionPolicy :: UnsupportedVersionPolicy } deriving (A.ToJSON, A.FromJSON) via (Schema (PayloadBundle c)) - deriving newtype (Semigroup) + deriving stock (Eq, Show) + +instance Semigroup (PayloadBundle c) where + bundle1 <> bundle2 = + PayloadBundle + { notifications = bundle1.notifications <> bundle2.notifications, + unsupportedVersionPolicy = bundle1.unsupportedVersionPolicy <> bundle2.unsupportedVersionPolicy + } instance (Typeable c) => ToSchema (PayloadBundle c) where schema = object $ PayloadBundle <$> notifications .= field "notifications" (nonEmptyArray schema) + <*> unsupportedVersionPolicy + .= fmap + (fromMaybe KeepQueued) + (optField "unsupportedVersionPolicy" schema) toBundle :: forall {k} (tag :: k). @@ -130,7 +166,10 @@ toBundle :: PayloadBundle (NotificationComponent k) toBundle reqId originDomain payload = let notif = fedNotifToBackendNotif @tag reqId originDomain payload - in PayloadBundle . pure $ notif + in PayloadBundle + { notifications = pure notif, + unsupportedVersionPolicy = KeepQueued + } makeBundle :: forall {k} (tag :: k) c. diff --git a/libs/wire-api-federation/test/Test/Wire/API/Federation/API/BackendNotificationsSpec.hs b/libs/wire-api-federation/test/Test/Wire/API/Federation/API/BackendNotificationsSpec.hs new file mode 100644 index 00000000000..2bb12c26503 --- /dev/null +++ b/libs/wire-api-federation/test/Test/Wire/API/Federation/API/BackendNotificationsSpec.hs @@ -0,0 +1,28 @@ +-- This file is part of the Wire Server implementation. +-- +-- Copyright (C) 2026 Wire Swiss GmbH +-- +-- This program is free software: you can redistribute it and/or modify it under +-- the terms of the GNU Affero General Public License as published by the Free +-- Software Foundation, either version 3 of the License, or (at your option) +-- any later version. +-- +-- This program is distributed in the hope that it will be useful, but WITHOUT +-- ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS +-- FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more +-- details. +-- +-- You should have received a copy of the GNU Affero General Public License along +-- with this program. If not, see . + +module Test.Wire.API.Federation.API.BackendNotificationsSpec where + +import Imports +import Test.Hspec +import Test.Wire.API.Federation.API.Util (jsonRoundTrip) +import Wire.API.Federation.BackendNotifications (UnsupportedVersionPolicy (..)) + +spec :: Spec +spec = describe "UnsupportedVersionPolicy" $ do + describe "roundtrip" $ do + jsonRoundTrip @UnsupportedVersionPolicy diff --git a/libs/wire-api-federation/test/Test/Wire/API/Federation/Golden/GoldenSpec.hs b/libs/wire-api-federation/test/Test/Wire/API/Federation/Golden/GoldenSpec.hs index 038f98b0d1e..80be3edb0a1 100644 --- a/libs/wire-api-federation/test/Test/Wire/API/Federation/Golden/GoldenSpec.hs +++ b/libs/wire-api-federation/test/Test/Wire/API/Federation/Golden/GoldenSpec.hs @@ -29,6 +29,7 @@ import Test.Wire.API.Federation.Golden.MessageSendResponse qualified as MessageS import Test.Wire.API.Federation.Golden.NewConnectionRequest qualified as NewConnectionRequest import Test.Wire.API.Federation.Golden.NewConnectionResponse qualified as NewConnectionResponse import Test.Wire.API.Federation.Golden.Runner (testObjects) +import Test.Wire.API.Federation.Golden.UnsupportedVersionPolicy qualified as UnsupportedVersionPolicy spec :: Spec spec = @@ -85,3 +86,7 @@ spec = (GetOne2OneConversationResponse.testObject_GetOne2OneConversationResponseBackendMismatch, "testObject_GetOne2OneConversationResponseBackendMismatch.json"), (GetOne2OneConversationResponse.testObject_GetOne2OneConversationResponseNotConnected, "testObject_GetOne2OneConversationResponseNotConnected.json") ] + testObjects + [ (UnsupportedVersionPolicy.testObjectUnsupportedVersionPolicyKeepQueued, "testObject_UnsupportedVersionPolicy_KeepQueued.json"), + (UnsupportedVersionPolicy.testObjectUnsupportedVersionPolicyDropIfUnsupported, "testObject_UnsupportedVersionPolicy_DropIfUnsupported.json") + ] diff --git a/libs/wire-api-federation/test/Test/Wire/API/Federation/Golden/UnsupportedVersionPolicy.hs b/libs/wire-api-federation/test/Test/Wire/API/Federation/Golden/UnsupportedVersionPolicy.hs new file mode 100644 index 00000000000..bd868c808d7 --- /dev/null +++ b/libs/wire-api-federation/test/Test/Wire/API/Federation/Golden/UnsupportedVersionPolicy.hs @@ -0,0 +1,26 @@ +-- This file is part of the Wire Server implementation. +-- +-- Copyright (C) 2026 Wire Swiss GmbH +-- +-- This program is free software: you can redistribute it and/or modify it under +-- the terms of the GNU Affero General Public License as published by the Free +-- Software Foundation, either version 3 of the License, or (at your option) any +-- later version. +-- +-- This program is distributed in the hope that it will be useful, but WITHOUT +-- ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS +-- FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more +-- details. +-- +-- You should have received a copy of the GNU Affero General Public License along +-- with this program. If not, see . + +module Test.Wire.API.Federation.Golden.UnsupportedVersionPolicy where + +import Wire.API.Federation.BackendNotifications (UnsupportedVersionPolicy (..)) + +testObjectUnsupportedVersionPolicyKeepQueued :: UnsupportedVersionPolicy +testObjectUnsupportedVersionPolicyKeepQueued = KeepQueued + +testObjectUnsupportedVersionPolicyDropIfUnsupported :: UnsupportedVersionPolicy +testObjectUnsupportedVersionPolicyDropIfUnsupported = DropIfUnsupported diff --git a/libs/wire-api-federation/test/golden/testObject_UnsupportedVersionPolicy_DropIfUnsupported.json b/libs/wire-api-federation/test/golden/testObject_UnsupportedVersionPolicy_DropIfUnsupported.json new file mode 100644 index 00000000000..ae02d02074a --- /dev/null +++ b/libs/wire-api-federation/test/golden/testObject_UnsupportedVersionPolicy_DropIfUnsupported.json @@ -0,0 +1 @@ +"drop_if_unsupported" diff --git a/libs/wire-api-federation/test/golden/testObject_UnsupportedVersionPolicy_KeepQueued.json b/libs/wire-api-federation/test/golden/testObject_UnsupportedVersionPolicy_KeepQueued.json new file mode 100644 index 00000000000..82b45e1fb25 --- /dev/null +++ b/libs/wire-api-federation/test/golden/testObject_UnsupportedVersionPolicy_KeepQueued.json @@ -0,0 +1 @@ +"keep_queued" diff --git a/libs/wire-api-federation/wire-api-federation.cabal b/libs/wire-api-federation/wire-api-federation.cabal index a451e2f01c0..5bb39296c38 100644 --- a/libs/wire-api-federation/wire-api-federation.cabal +++ b/libs/wire-api-federation/wire-api-federation.cabal @@ -135,6 +135,7 @@ test-suite spec -- cabal-fmt: expand test other-modules: Main + Test.Wire.API.Federation.API.BackendNotificationsSpec Test.Wire.API.Federation.API.BrigSpec Test.Wire.API.Federation.API.GalleySpec Test.Wire.API.Federation.API.Util @@ -149,6 +150,7 @@ test-suite spec Test.Wire.API.Federation.Golden.NewConnectionRequest Test.Wire.API.Federation.Golden.NewConnectionResponse Test.Wire.API.Federation.Golden.Runner + Test.Wire.API.Federation.Golden.UnsupportedVersionPolicy hs-source-dirs: test default-extensions: diff --git a/services/background-worker/src/Wire/BackendNotificationPusher.hs b/services/background-worker/src/Wire/BackendNotificationPusher.hs index cadce5c0270..0654dbd3344 100644 --- a/services/background-worker/src/Wire/BackendNotificationPusher.hs +++ b/services/background-worker/src/Wire/BackendNotificationPusher.hs @@ -207,10 +207,23 @@ pushNotification runningFlag targetDomain (msg, envelope) = do -- compute the best usable version in a notification let bestVersion = bodyVersions >=> flip latestCommonVersion remoteVersions case pairedMaximumOn bestVersion (toList (notifications bundle)) of - (_, Nothing) -> - Log.fatal $ - Log.msg (Log.val "No federation API version in common, the notification will be ignored") - . Log.field "domain" (domainText targetDomain) + (_, Nothing) -> do + metrics <- asks backendNotificationMetrics + case bundle.unsupportedVersionPolicy of + KeepQueued -> do + Log.fatal $ + Log.msg (Log.val "No federation API version in common; the notification will remain queued") + . Log.field "domain" (domainText targetDomain) + . Log.field "paths" (Text.intercalate "," $ map (.path) $ toList bundle.notifications) + withLabel metrics.stuckQueuesGauge (domainText targetDomain) (flip setGauge 1) + DropIfUnsupported -> do + Log.warn $ + Log.msg (Log.val "Dropping notification because the target backend supports no compatible federation API version") + . Log.field "domain" (domainText targetDomain) + . Log.field "paths" (Text.intercalate "," $ map (.path) $ toList bundle.notifications) + lift $ ack envelope + withLabel metrics.droppedUnsupportedVersionCounter (domainText targetDomain) incCounter + withLabel metrics.stuckQueuesGauge (domainText targetDomain) (flip setGauge 0) (notif, cveVersion) -> do ceFederator <- asks (.federatorInternal) ceHttp2Manager <- asks http2Manager diff --git a/services/background-worker/src/Wire/BackgroundWorker/Env.hs b/services/background-worker/src/Wire/BackgroundWorker/Env.hs index ed784a33db9..10886fd8332 100644 --- a/services/background-worker/src/Wire/BackgroundWorker/Env.hs +++ b/services/background-worker/src/Wire/BackgroundWorker/Env.hs @@ -118,6 +118,7 @@ data Env = Env data BackendNotificationMetrics = BackendNotificationMetrics { pushedCounter :: Vector Text Counter, + droppedUnsupportedVersionCounter :: Vector Text Counter, errorCounter :: Vector Text Counter, stuckQueuesGauge :: Vector Text Gauge } @@ -130,6 +131,7 @@ mkBackendNotificationMetrics :: IO BackendNotificationMetrics mkBackendNotificationMetrics = BackendNotificationMetrics <$> register (vector "targetDomain" $ counter $ Prometheus.Info "wire_backend_notifications_pushed" "Number of notifications pushed") + <*> register (vector "targetDomain" $ counter $ Prometheus.Info "wire_backend_notifications_dropped_unsupported_version" "Number of notifications dropped because the target backend supports no compatible federation API version") <*> 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") diff --git a/services/background-worker/test/Test/Wire/BackendNotificationPusherSpec.hs b/services/background-worker/test/Test/Wire/BackendNotificationPusherSpec.hs index 7222120d93a..be46a03c648 100644 --- a/services/background-worker/test/Test/Wire/BackendNotificationPusherSpec.hs +++ b/services/background-worker/test/Test/Wire/BackendNotificationPusherSpec.hs @@ -66,6 +66,7 @@ import Wire.API.Federation.API.Brig import Wire.API.Federation.API.Common import Wire.API.Federation.API.Galley import Wire.API.Federation.BackendNotifications +import Wire.API.Federation.Version import Wire.API.RawJson import Wire.API.Team.FeatureFlags import Wire.BackendNotificationPusher @@ -77,6 +78,17 @@ import Wire.RateLimit.Interpreter (newRateLimitEnv) spec :: Spec spec = do + describe "PayloadBundle" $ do + it "should default to keeping a notification queued when decoding a bundle without a policy" $ do + let bundle = testBundle KeepQueued (rangeFromVersion V1) + oldBundle = Aeson.object ["notifications" .= bundle.notifications] + Aeson.fromJSON @(PayloadBundle 'Brig) oldBundle `shouldBe` Aeson.Success bundle + + it "should use the safe keep-queued policy when combining bundles with different policies" $ do + let keepQueuedBundle = testBundle KeepQueued (rangeFromVersion V1) + dropBundle = testBundle DropIfUnsupported (rangeFromVersion V1) + (keepQueuedBundle <> dropBundle).unsupportedVersionPolicy `shouldBe` KeepQueued + describe "pushNotification" $ do it "should push notifications" $ do let origDomain = Domain "origin.example.com" @@ -253,6 +265,119 @@ spec = do getVectorWith env.backendNotificationMetrics.pushedCounter getCounter `shouldReturn` [(domainText targetDomain, 1)] + it "should leave an unsupported notification queued by default" $ do + let targetDomain = Domain "target.example.com" + bundle = testBundle KeepQueued (rangeFromVersion V1) + envelope <- newMockEnvelope + let msg = + Q.newMsg + { Q.msgBody = Aeson.encode bundle, + Q.msgContentType = Just "application/json" + } + runningFlag <- newMVar () + (env, fedReqs) <- + withTempMockFederator def {versions = [0]} . runTestAppT $ do + wait =<< pushNotification runningFlag targetDomain (msg, envelope) + ask + + readIORef envelope.acks `shouldReturn` 0 + readIORef envelope.rejections `shouldReturn` [] + fedReqs `shouldBe` [] + getVectorWith env.backendNotificationMetrics.droppedUnsupportedVersionCounter getCounter + `shouldReturn` [] + getVectorWith env.backendNotificationMetrics.stuckQueuesGauge getGauge + `shouldReturn` [(domainText targetDomain, 1)] + + it "should drop an unsupported notification when configured" $ do + let targetDomain = Domain "target.example.com" + bundle = testBundle DropIfUnsupported (rangeFromVersion V1) + envelope <- newMockEnvelope + let msg = + Q.newMsg + { Q.msgBody = Aeson.encode bundle, + Q.msgContentType = Just "application/json" + } + runningFlag <- newMVar () + (env, fedReqs) <- + withTempMockFederator def {versions = [0]} . runTestAppT $ do + wait =<< pushNotification runningFlag targetDomain (msg, envelope) + ask + + readIORef envelope.acks `shouldReturn` 1 + readIORef envelope.rejections `shouldReturn` [] + fedReqs `shouldBe` [] + getVectorWith env.backendNotificationMetrics.droppedUnsupportedVersionCounter getCounter + `shouldReturn` [(domainText targetDomain, 1)] + getVectorWith env.backendNotificationMetrics.stuckQueuesGauge getGauge + `shouldReturn` [(domainText targetDomain, 0)] + + it "should deliver a compatible notification with the drop policy" $ do + let targetDomain = Domain "target.example.com" + bundle = testBundle DropIfUnsupported (rangeFromVersion V1) + envelope <- newMockEnvelope + let msg = + Q.newMsg + { Q.msgBody = Aeson.encode bundle, + Q.msgContentType = Just "application/json" + } + runningFlag <- newMVar () + (env, fedReqs) <- + withTempMockFederator def {versions = [1]} . runTestAppT $ do + wait =<< pushNotification runningFlag targetDomain (msg, envelope) + ask + + readIORef envelope.acks `shouldReturn` 1 + readIORef envelope.rejections `shouldReturn` [] + fedReqs + `shouldBe` [ FederatedRequest + { frTargetDomain = targetDomain, + frOriginDomain = testOriginDomain, + frComponent = Brig, + frRPC = "unsupported-version-test", + frBody = Aeson.encode testNotificationBody + } + ] + getVectorWith env.backendNotificationMetrics.droppedUnsupportedVersionCounter getCounter + `shouldReturn` [] + + it "should retry delivery failures instead of applying the drop policy" $ do + isRemoteBrokenRef <- newIORef True + fedCalls <- newIORef (0 :: Int) + let mockRemote :: req -> IO MockResponse + mockRemote _ = do + isRemoteBroken <- readIORef isRemoteBrokenRef + atomicModifyIORef fedCalls $ \c -> (c + 1, ()) + pure $ + if isRemoteBroken + then MockResponse status200 "text/html" "down for maintenance" + else MockResponse status200 "application/json" (Aeson.encode EmptyResponse) + targetDomain = Domain "target.example.com" + bundle = testBundle DropIfUnsupported (rangeFromVersion V1) + envelope <- newMockEnvelope + let msg = + Q.newMsg + { Q.msgBody = Aeson.encode bundle, + Q.msgContentType = Just "application/json" + } + runningFlag <- newMVar () + env <- testEnv + pushThread <- + async $ withTempMockFederator def {handler = mockRemote, versions = [1]} . runTestAppTWithEnv env $ do + wait =<< pushNotification runningFlag targetDomain (msg, envelope) + + untilM $ (>= 2) <$> readIORef fedCalls + readIORef envelope.acks `shouldReturn` 0 + getVectorWith env.backendNotificationMetrics.droppedUnsupportedVersionCounter getCounter + `shouldReturn` [] + + writeIORef isRemoteBrokenRef False + void $ wait pushThread + + readIORef envelope.acks `shouldReturn` 1 + readIORef envelope.rejections `shouldReturn` [] + getVectorWith env.backendNotificationMetrics.droppedUnsupportedVersionCounter getCounter + `shouldReturn` [] + it "should reject invalid notifications" $ do envelope <- newMockEnvelope let msg = @@ -487,6 +612,28 @@ spec = do calls `shouldSatisfy` (\c -> length c >= 2) mapM_ (\vhost -> vhost `shouldBe` rabbitmqVHost) calls +testOriginDomain :: Domain +testOriginDomain = Domain "origin.example.com" + +testNotificationBody :: Aeson.Value +testNotificationBody = Aeson.object ["foo" .= ("bar" :: Text)] + +testBundle :: UnsupportedVersionPolicy -> VersionRange -> PayloadBundle 'Brig +testBundle policy versions = + PayloadBundle + { notifications = + pure + BackendNotification + { targetComponent = Brig, + ownDomain = testOriginDomain, + path = "/unsupported-version-test", + body = RawJson $ Aeson.encode testNotificationBody, + bodyVersions = Just versions, + requestId = Just $ RequestId defRequestId + }, + unsupportedVersionPolicy = policy + } + untilM :: (Monad m) => m Bool -> m () untilM action = do b <- action