WPB-28421 add an opt in policy for dropping unsupported federated notifications

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.

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.

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 <opensource@wire.com>
+--
+-- This program is free software: you can redistribute it and/or modify it under
+-- the terms of the GNU Affero General Public License as published by the Free
+-- Software Foundation, either version 3 of the License, or (at your option)
+-- any later version.
+--
+-- This program is distributed in the hope that it will be useful, but WITHOUT
+-- ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+-- FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more
+-- details.
+--
+-- You should have received a copy of the GNU Affero General Public License along
+-- with this program. If not, see <https://www.gnu.org/licenses/>.
+
+module 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

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")
+      ]

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 <opensource@wire.com>
+--
+-- This program is free software: you can redistribute it and/or modify it under
+-- the terms of the GNU Affero General Public License as published by the Free
+-- Software Foundation, either version 3 of the License, or (at your option) any
+-- later version.
+--
+-- This program is distributed in the hope that it will be useful, but WITHOUT
+-- ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+-- FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more
+-- details.
+--
+-- You should have received a copy of the GNU Affero General Public License along
+-- with this program. If not, see <https://www.gnu.org/licenses/>.
+
+module Test.Wire.API.Federation.Golden.UnsupportedVersionPolicy where
+
+import Wire.API.Federation.BackendNotifications (UnsupportedVersionPolicy (..))
+
+testObjectUnsupportedVersionPolicyKeepQueued :: UnsupportedVersionPolicy
+testObjectUnsupportedVersionPolicyKeepQueued = KeepQueued
+
+testObjectUnsupportedVersionPolicyDropIfUnsupported :: UnsupportedVersionPolicy
+testObjectUnsupportedVersionPolicyDropIfUnsupported = DropIfUnsupported

libs/wire-api-federation/test/golden/testObject_UnsupportedVersionPolicy_DropIfUnsupported.json

@@ -0,0 +1 @@
+"drop_if_unsupported"

libs/wire-api-federation/test/golden/testObject_UnsupportedVersionPolicy_KeepQueued.json

@@ -0,0 +1 @@
+"keep_queued"

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:

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

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")
 

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" "<marquee>down for maintenance</marquee>"
+                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