add test, treat BAD_SERVICE as temp error, only remove queue associations on service errors

Author: epoberezkin

src/Simplex/Messaging/Agent.hs

@@ -1479,10 +1479,12 @@ subscribeAllConnections' c onlyNeeded activeUserId_ = handleErr $ do
     subscribeService useServices us@(userId, srv) = fmap ((us,) . fromRight False) $ tryAllErrors' $ do
       withStore' c (\db -> getSubscriptionService db userId srv) >>= \case
         Just serviceSub -> case M.lookup userId useServices of
-          -- TODO [certs rcv] improve logic to differentiate between permanent and temporary service subscription errors,
-          -- as the current logic would fall back to per-queue subscriptions on ANY service subscription error (e.g., network connection error).
-          Just True -> isRight <$> tryAllErrors (subscribeClientService c True userId srv serviceSub)
-          _ -> False <$ withStore' c (\db -> unassocUserServerRcvQueueSubs db userId srv)
+          Just True -> tryAllErrors (subscribeClientService c True userId srv serviceSub) >>= \case
+            Left e | clientServiceError e -> unassocQueues $> False
+            _ -> pure True
+          _ -> unassocQueues $> False
+          where
+            unassocQueues = withStore' c $ \db -> unassocUserServerRcvQueueSubs db userId srv
         _ -> pure False
     subscribeUserServer :: Int -> TVar Int -> ((UserId, SMPServer), ServiceAssoc) -> AM' (Either AgentErrorType Int)
     subscribeUserServer maxPending currPending ((userId, srv), hasService) = do
@@ -1498,7 +1500,6 @@ subscribeAllConnections' c onlyNeeded activeUserId_ = handleErr $ do
       where
         subscribe qs = do
           rs <- subscribeUserServerQueues c userId srv qs
-          -- TODO [certs rcv] storeClientServiceAssocs store associations of queues with client service ID
           ns <- asks ntfSupervisor
           whenM (liftIO $ hasInstantNotifications ns) $ sendNtfCreate ns rs
     sendNtfCreate :: NtfSupervisor -> [(RcvQueueSub, Either AgentErrorType (Maybe SMP.ServiceId))] -> AM' ()

src/Simplex/Messaging/Agent/Client.hs

@@ -120,6 +120,7 @@ module Simplex.Messaging.Agent.Client
     getAgentSubscriptions,
     slowNetworkConfig,
     protocolClientError,
+    clientServiceError,
     Worker (..),
     SessionVar (..),
     SubscriptionsInfo (..),
@@ -1260,7 +1261,6 @@ protocolClientError protocolError_ host = \case
 clientServiceError :: AgentErrorType -> Bool
 clientServiceError = \case
   BROKER _ NO_SERVICE -> True
-  BROKER _ (TRANSPORT (TEHandshake BAD_SERVICE)) -> True -- TODO [certs rcv] this error may be temporary, so we should possibly resubscribe.
   SMP _ SMP.SERVICE -> True
   SMP _ (SMP.PROXY (SMP.BROKER NO_SERVICE)) -> True -- for completeness, it cannot happen.
   _ -> False
@@ -1566,6 +1566,8 @@ temporaryAgentError :: AgentErrorType -> Bool
 temporaryAgentError = \case
   BROKER _ e -> tempBrokerError e
   SMP _ (SMP.PROXY (SMP.BROKER e)) -> tempBrokerError e
+  SMP _ (SMP.STORE _) -> True
+  NTF _ (SMP.STORE _) -> True
   XFTP _ XFTP.TIMEOUT -> True
   PROXY _ _ (ProxyProtocolError (SMP.PROXY (SMP.BROKER e))) -> tempBrokerError e
   PROXY _ _ (ProxyProtocolError (SMP.PROXY SMP.NO_SESSION)) -> True
@@ -1576,6 +1578,7 @@ temporaryAgentError = \case
     tempBrokerError = \case
       NETWORK _ -> True
       TIMEOUT -> True
+      TRANSPORT (TEHandshake BAD_SERVICE) -> True -- this error is considered temporary because it is DB error
       _ -> False
 
 temporaryOrHostError :: AgentErrorType -> Bool
@@ -1722,8 +1725,12 @@ processClientNotices c@AgentClient {presetServers} tSess notices = do
       notifySub' c "" $ ERR e
 
 resubscribeClientService :: AgentClient -> SMPTransportSession -> ServiceSub -> AM ServiceSubResult
-resubscribeClientService c tSess serviceSub =
-  withServiceClient c tSess $ \smp _ -> subscribeClientService_ c True tSess smp serviceSub
+resubscribeClientService c tSess@(userId, srv, _) serviceSub =
+  withServiceClient c tSess (\smp _ -> subscribeClientService_ c True tSess smp serviceSub) `catchE` \e -> do
+    when (clientServiceError e) $ do
+      qs <- withStore' c $ \db -> unassocUserServerRcvQueueSubs db userId srv
+      void $ lift $ subscribeUserServerQueues c userId srv qs
+    throwE e
 
 -- TODO [certs rcv] update service in the database if it has different ID and re-associate queues, and send event
 subscribeClientService :: AgentClient -> Bool -> UserId -> SMPServer -> ServiceSub -> AM ServiceSubResult
@@ -1737,16 +1744,10 @@ subscribeClientService c withEvent userId srv (ServiceSub _ n idsHash) =
 
 withServiceClient :: AgentClient -> SMPTransportSession -> (SMPClient -> ServiceId -> ExceptT SMPClientError IO a) -> AM a
 withServiceClient c tSess@(userId, srv, _) subscribe =
-  unassocOnError $ withLogClient c NRMBackground tSess B.empty "SUBS" $ \(SMPConnectedClient smp _) ->
+  withLogClient c NRMBackground tSess B.empty "SUBS" $ \(SMPConnectedClient smp _) ->
     case (\THClientService {serviceId} -> serviceId) <$> smpClientService smp of
       Just smpServiceId -> subscribe smp smpServiceId
       Nothing -> throwE PCEServiceUnavailable
-  where
-    unassocOnError a = a `catchE` \e -> do
-      when (clientServiceError e) $ do
-        qs <- withStore' c $ \db -> unassocUserServerRcvQueueSubs db userId srv
-        void $ lift $ subscribeUserServerQueues c userId srv qs
-      throwE e
 
 -- TODO [certs rcv] send subscription error event?
 subscribeClientService_ :: AgentClient -> Bool -> SMPTransportSession -> SMPClient -> ServiceSub -> ExceptT SMPClientError IO ServiceSubResult

src/Simplex/Messaging/Client.hs

@@ -782,7 +782,6 @@ temporaryClientError = \case
 smpClientServiceError :: SMPClientError -> Bool
 smpClientServiceError = \case
   PCEServiceUnavailable -> True
-  PCETransportError (TEHandshake BAD_SERVICE) -> True -- TODO [certs rcv] this error may be temporary, so we should possibly resubscribe.
   PCEProtocolError SERVICE -> True
   PCEProtocolError (PROXY (BROKER NO_SERVICE)) -> True -- for completeness, it cannot happen.
   _ -> False

src/Simplex/Messaging/Protocol.hs

@@ -143,6 +143,7 @@ module Simplex.Messaging.Protocol
     IdsHash (..),
     ServiceSub (..),
     ServiceSubResult (..),
+    ServiceSubError (..),
     serviceSubResult,
     queueIdsHash,
     queueIdHash,

tests/AgentTests/FunctionalAPITests.hs

@@ -66,7 +66,7 @@ import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Either (isRight)
 import Data.Int (Int64)
-import Data.List (find, isSuffixOf, nub)
+import Data.List (find, isPrefixOf, isSuffixOf, nub)
 import Data.List.NonEmpty (NonEmpty)
 import qualified Data.Map as M
 import Data.Maybe (isJust, isNothing)
@@ -113,7 +113,7 @@ import Simplex.Messaging.Util (bshow, diffToMicroseconds)
 import Simplex.Messaging.Version (VersionRange (..))
 import qualified Simplex.Messaging.Version as V
 import Simplex.Messaging.Version.Internal (Version (..))
-import System.Directory (copyFile, renameFile)
+import System.Directory (copyFile, removeFile, renameFile)
 import Test.Hspec hiding (fit, it)
 import UnliftIO
 import Util
@@ -124,10 +124,13 @@ import Fixtures
 #endif
 #if defined(dbServerPostgres)
 import qualified Database.PostgreSQL.Simple as PSQL
+import qualified Simplex.Messaging.Agent.Store.Postgres as Postgres
+import qualified Simplex.Messaging.Agent.Store.Postgres.Common as Postgres
 import Simplex.Messaging.Server.MsgStore.Journal (JournalQueue)
 import Simplex.Messaging.Server.MsgStore.Postgres (PostgresQueue)
 import Simplex.Messaging.Server.MsgStore.Types (QSType (..))
 import Simplex.Messaging.Server.QueueStore.Postgres
+import Simplex.Messaging.Server.QueueStore.Postgres.Migrations
 import Simplex.Messaging.Server.QueueStore.Types (QueueStoreClass (..))
 #endif
 
@@ -476,6 +479,7 @@ functionalAPITests ps = do
       withSmpServer ps testTwoUsers
   describe "Client service certificates" $ do
     it "should connect, subscribe and reconnect as a service" $ testClientServiceConnection ps
+    it "should re-subscribe when service ID changed" $ testClientServiceIDChange ps
   describe "Connection switch" $ do
     describe "should switch delivery to the new queue" $
       testServerMatrix2 ps testSwitchConnection
@@ -3715,6 +3719,47 @@ testClientServiceConnection ps = do
       exchangeGreetingsMsgId 4 user sId' service uId'
       exchangeGreetingsMsgId 10 service uId user sId
 
+testClientServiceIDChange :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
+testClientServiceIDChange ps@(_, ASType qs _) = do
+  (sId, uId) <- withAgentClientsServers2 (agentCfg, initAgentServersClientService) (agentCfg, initAgentServers) $ \service user -> do
+    withSmpServerStoreLogOn ps testPort $ \_ -> runRight $ do
+      conns@(sId, uId) <- makeConnection service user
+      exchangeGreetings service uId user sId
+      pure conns
+  _ :: () <- case qs of
+    SQSPostgres -> do
+#if defined(dbServerPostgres)
+      st <- either (error . show) pure =<< Postgres.createDBStore testStoreDBOpts serverMigrations (MigrationConfig MCError Nothing)
+      void $ Postgres.withTransaction st (`PSQL.execute_` "DELETE FROM services")
+#else
+      pure ()
+#endif
+    SQSMemory -> do
+      s <- readFile testStoreLogFile
+      removeFile testStoreLogFile
+      writeFile testStoreLogFile $ unlines $ filter (not . ("NEW_SERVICE" `isPrefixOf`)) $ lines s
+  withAgentClientsServers2 (agentCfg, initAgentServersClientService) (agentCfg, initAgentServers) $ \service user -> do
+    withSmpServerStoreLogOn ps testPort $ \_ -> runRight $ do
+      subscribeAllConnections service False Nothing
+      liftIO $ getInAnyOrder service
+        [ \case ("", "", AEvt SAENone (SERVICE_UP _ (SMP.ServiceSubResult (Just (SMP.SSErrorQueueCount 1 0)) (SMP.ServiceSub _ 0 _)))) -> True; _ -> False,
+          \case ("", "", AEvt SAENone (SERVICE_ALL _)) -> True; _ -> False,
+          \case ("", "", AEvt SAENone (UP _ _)) -> True; _ -> False
+        ]
+      subscribeAllConnections user False Nothing
+      ("", "", UP _ [_]) <- nGet user
+      exchangeGreetingsMsgId 4 service uId user sId
+  -- disable service in the client
+  -- The test uses True for non-existing user to make sure it's removed for user 1,
+  -- because if no users use services, then it won't be checking them to optimize for most clients.
+  withAgentClientsServers2 (agentCfg, initAgentServers {useServices = M.fromList [(100, True)]}) (agentCfg, initAgentServers) $ \notService user -> do
+    withSmpServerStoreLogOn ps testPort $ \_ -> runRight $ do
+      subscribeAllConnections notService False Nothing
+      ("", "", UP _ [_]) <- nGet notService
+      subscribeAllConnections user False Nothing
+      ("", "", UP _ [_]) <- nGet user
+      exchangeGreetingsMsgId 6 notService uId user sId
+
 getSMPAgentClient' :: Int -> AgentConfig -> InitialAgentServers -> String -> IO AgentClient
 getSMPAgentClient' clientId cfg' initServers dbPath = do
   Right st <- liftIO $ createStore dbPath