fixing services WIP

Author: epoberezkin

src/Simplex/Messaging/Agent.hs

@@ -1453,22 +1453,18 @@ subscribeAllConnections' c onlyNeeded activeUserId_ = handleErr $ do
           Just activeUserId -> sortOn (\(uId, _) -> if uId == activeUserId then 0 else 1 :: Int) userSrvs
           Nothing -> userSrvs
     useServices <- readTVarIO $ useClientServices c
-    -- These options are possible below:
-    -- 1) services fully disabled:
-    --    No service subscriptions will be attempted, and existing services and association will remain in in the database,
-    --    but they will be ignored because of hasService parameter set to False.
-    --    This approach preserves performance for all clients that do not use services.
-    -- 2) at least one user ID has services enabled:
-    --    Service will be loaded for all user/server combinations:
-    --    a) service is enabled for user ID and service record exists: subscription will be attempted,
-    --    b) service is disabled and record exists: service record and all associations will be removed,
-    --    c) service is disabled or no record: no subscription attempt.
+    -- Service will be loaded for all user/server combinations:
+    -- a) service is enabled for user ID and service record exists: subscription will be attempted,
+    -- b) service is disabled and record exists: service record and all associations will be removed,
+    -- c) service is disabled or no record: no subscription attempt.
     -- On successful service subscription, only unassociated queues will be subscribed.
-    userSrvs'' <-
-      if any id useServices
-        then lift $ mapConcurrently (subscribeService useServices) userSrvs'
-        else pure $ map (,False) userSrvs'
-    rs <- lift $ mapConcurrently (subscribeUserServer maxPending currPending) userSrvs''
+    when (clientId c == 1) $ liftIO $ print 11
+    userSrvs2 <- withStore' c $ \db -> mapM (getService db useServices) userSrvs'
+    when (clientId c == 1) $ liftIO $ print 22
+    userSrvs3 <- lift $ mapConcurrently subscribeService userSrvs2
+    when (clientId c == 1) $ liftIO $ print 33
+    rs <- lift $ mapConcurrently (subscribeUserServer maxPending currPending) userSrvs3
+    when (clientId c == 1) $ liftIO $ print 44
     let (errs, oks) = partitionEithers rs
     logInfo $ "subscribed " <> tshow (sum oks) <> " queues"
     forM_ (L.nonEmpty errs) $ notifySub c . ERRS . L.map ("",)
@@ -1477,23 +1473,27 @@ subscribeAllConnections' c onlyNeeded activeUserId_ = handleErr $ do
   resumeAllCommands c
   where
     handleErr = (`catchAllErrors` \e -> notifySub' c "" (ERR e) >> throwE e)
-    subscribeService :: Map UserId Bool -> (UserId, SMPServer) -> AM' ((UserId, SMPServer), ServiceAssoc)
-    subscribeService useServices us@(userId, srv) = fmap ((us,) . fromRight False) $ tryAllErrors' $ do
-      withStore' c (\db -> getSubscriptionService db userId srv) >>= \case
+    getService :: DB.Connection -> Map UserId Bool -> (UserId, SMPServer) -> IO ((UserId, SMPServer), Maybe ServiceSub)
+    getService db useServices us@(userId, srv) =
+      fmap (us,) $ getSubscriptionService db userId srv >>= \case
         Just serviceSub -> case M.lookup userId useServices of
-          Just True -> tryAllErrors (subscribeClientService c True userId srv serviceSub) >>= \case
-            Right (ServiceSubResult e _) -> case e of
-              Just SSErrorServiceId {} -> unassocQueues
-              -- Below would resubscribe all queues after service was disabled and re-enabled
-              -- Possibly, we should always resubscribe all with expected is greated than subscribed
-              Just SSErrorQueueCount {expectedQueueCount = n, subscribedQueueCount = n'} | n > 0 && n' == 0 -> unassocQueues
-              _ -> pure True
-            Left e -> do
-              atomically $ writeTBQueue (subQ c) ("", "", AEvt SAEConn $ ERR e)
-              if clientServiceError e
-                then unassocQueues
-                else pure True
-          _ -> unassocQueues
+          Just True -> pure $ Just serviceSub
+          _ -> Nothing <$ unassocUserServerRcvQueueSubs' db userId srv
+        _ -> pure Nothing
+    subscribeService :: ((UserId, SMPServer), Maybe ServiceSub) -> AM' ((UserId, SMPServer), ServiceAssoc)
+    subscribeService (us@(userId, srv), serviceSub_) = fmap ((us,) . fromRight False) $ tryAllErrors' $
+      case serviceSub_ of
+        Just serviceSub -> tryAllErrors (subscribeClientService c True userId srv serviceSub) >>= \case
+          Right (ServiceSubResult e _) -> case e of
+            Just SSErrorServiceId {} -> unassocQueues
+            -- Possibly, we should always resubscribe all when expected is greater than subscribed
+            Just SSErrorQueueCount {expectedQueueCount = n, subscribedQueueCount = n'} | n > 0 && n' == 0 -> unassocQueues
+            _ -> pure True
+          Left e -> do
+            atomically $ writeTBQueue (subQ c) ("", "", AEvt SAEConn $ ERR e)
+            if clientServiceError e
+              then unassocQueues
+              else pure True
           where
             unassocQueues :: AM Bool
             unassocQueues = False <$ withStore' c (\db -> unassocUserServerRcvQueueSubs' db userId srv)

src/Simplex/Messaging/Agent/Client.hs

@@ -743,13 +743,22 @@ smpConnectClient c@AgentClient {smpClients, msgQ, proxySessTs, presetDomains} nm
       cfg <- lift $ getClientConfig c smpCfg
       g <- asks random
       service <- getServiceCredentials c userId srv
+      when (clientId c == 1) $ liftIO $ print $ snd <$> service
       let cfg' = cfg {serviceCredentials = fst <$> service}
       env <- ask
-      smp <- liftError (protocolClientError SMP $ B.unpack $ strEncode srv) $ do
+      when (clientId c == 1) $ liftIO $ print "before getProtocolClient"
+      smp <- liftError (protocolClientError SMP $ B.unpack $ strEncode srv) $ ExceptT $ do
         ts <- readTVarIO proxySessTs
-        ExceptT $ getProtocolClient g nm tSess cfg' presetDomains (Just msgQ) ts $ smpClientDisconnected c tSess env v' prs
+        r <- getProtocolClient g nm tSess cfg' presetDomains (Just msgQ) ts $ smpClientDisconnected c tSess env v' prs
+        case r of
+          Left e -> when (clientId c == 1) $ liftIO $ print e
+          Right _ -> pure ()
+        pure r
+      when (clientId c == 1) $ liftIO $ print "after getProtocolClient"
       atomically $ SS.setSessionId tSess (sessionId $ thParams smp) $ currentSubs c
+      when (clientId c == 1) $ liftIO $ print "before updateClientService"
       updateClientService service smp
+      when (clientId c == 1) $ liftIO $ print "after updateClientService"
       pure SMPConnectedClient {connectedClient = smp, proxiedRelays = prs}
     updateClientService service smp = case (service, smpClientService smp) of
       (Just (_, serviceId_), Just THClientService {serviceId}) -> withStore' c $ \db -> do

src/Simplex/Messaging/Agent/Store/AgentStore.hs

@@ -38,7 +38,6 @@ module Simplex.Messaging.Agent.Store.AgentStore
     -- * Client services
     createClientService,
     getClientServiceCredentials,
-    getSubscriptionServices,
     getSubscriptionService,
     getClientServiceServers,
     setClientServiceId,
@@ -345,7 +344,7 @@ handleSQLError err e = case constraintViolation e of
 handleSQLError :: StoreError -> SQLError -> StoreError
 handleSQLError err e
   | SQL.sqlError e == SQL.ErrorConstraint = err
-  | otherwise = SEInternal $ bshow e
+  | otherwise = SEInternal $ encodeUtf8 $ tshow e <> ": " <> SQL.sqlErrorDetails e <> ", " <> SQL.sqlErrorContext e
 #endif
 
 createUserRecord :: DB.Connection -> IO UserId
@@ -409,6 +408,7 @@ deleteUsersWithoutConns db = do
 createClientService :: DB.Connection -> UserId -> SMPServer -> (C.KeyHash, TLS.Credential) -> IO ()
 createClientService db userId srv (kh, (cert, pk)) = do
   serverKeyHash_ <- createServer_ db srv
+  print "createClientService"
   DB.execute
     db
     [sql|
@@ -440,44 +440,44 @@ getClientServiceCredentials db userId srv =
   where
     toService (kh, cert, pk, serviceId_) = ((kh, (cert, pk)), serviceId_)
 
-getSubscriptionServices :: DB.Connection -> IO [(UserId, (SMPServer, ServiceSub))]
-getSubscriptionServices db = map toUserService <$> DB.query_ db clientServiceQuery
-  where
-    toUserService (Only userId :. serviceRow) = (userId, toServerService serviceRow)
-
 getSubscriptionService :: DB.Connection -> UserId -> SMPServer -> IO (Maybe ServiceSub)
-getSubscriptionService db userId (SMPServer h p kh) =
-  maybeFirstRow toService $
+getSubscriptionService db userId (SMPServer h p kh) = do
+  print "1111 before getSubscriptionService"
+  r <- maybeFirstRow toService $
     DB.query
       db
       [sql|
         SELECT c.service_id, c.service_queue_count, c.service_queue_ids_hash
         FROM client_services c
         JOIN servers s ON s.host = c.host AND s.port = c.port
-        WHERE c.user_id = ? AND c.host = ? AND c.port = ? AND COALESCE(c.server_key_hash, s.key_hash) = ?
+        WHERE c.user_id = ? AND c.host = ? AND c.port = ? AND COALESCE(c.server_key_hash, s.key_hash) = ? AND service_id IS NOT NULL
       |]
       (userId, h, p, kh)
+  print "2222 after getSubscriptionService"
+  pure r
   where
     toService (serviceId, qCnt, idsHash) = ServiceSub serviceId qCnt idsHash
 
 getClientServiceServers :: DB.Connection -> UserId -> IO [(SMPServer, ServiceSub)]
 getClientServiceServers db userId =
-  map toServerService <$> DB.query db (clientServiceQuery <> " WHERE c.user_id = ?") (Only userId)
-
-clientServiceQuery :: Query
-clientServiceQuery =
-  [sql|
-    SELECT c.host, c.port, COALESCE(c.server_key_hash, s.key_hash), c.service_id, c.service_queue_count, c.service_queue_ids_hash
-    FROM client_services c
-    JOIN servers s ON s.host = c.host AND s.port = c.port
-  |]
+  map toServerService <$>
+    DB.query
+      db
+      [sql|
+        SELECT c.host, c.port, COALESCE(c.server_key_hash, s.key_hash), c.service_id, c.service_queue_count, c.service_queue_ids_hash
+        FROM client_services c
+        JOIN servers s ON s.host = c.host AND s.port = c.port
+        WHERE c.user_id = ? AND service_id IS NOT NULL
+      |]
+      (Only userId)
 
 toServerService :: (NonEmpty TransportHost, ServiceName, C.KeyHash, ServiceId, Int64, Binary ByteString) -> (ProtocolServer 'PSMP, ServiceSub)
 toServerService (host, port, kh, serviceId, n, Binary idsHash) =
   (SMPServer host port kh, ServiceSub serviceId n (IdsHash idsHash))
 
 setClientServiceId :: DB.Connection -> UserId -> SMPServer -> ServiceId -> IO ()
-setClientServiceId db userId srv serviceId =
+setClientServiceId db userId srv serviceId = do
+  print "setClientServiceId"
   DB.execute
     db
     [sql|
@@ -488,14 +488,20 @@ setClientServiceId db userId srv serviceId =
     (serviceId, userId, host srv, port srv)
 
 deleteClientService :: DB.Connection -> UserId -> SMPServer -> IO ()
-deleteClientService db userId srv =
+deleteClientService db userId (SMPServer h p kh) =
   DB.execute
     db
     [sql|
       DELETE FROM client_services
       WHERE user_id = ? AND host = ? AND port = ?
+        AND EXISTS (
+          SELECT 1 FROM servers s
+          WHERE s.host = client_services.host
+            AND s.port = client_services.port
+            AND COALESCE(client_services.server_key_hash, s.key_hash) = ?
+        );
     |]
-    (userId, host srv, port srv)
+    (userId, h, p, Just kh)
 
 deleteClientServices :: DB.Connection -> UserId -> IO ()
 deleteClientServices db userId = do
@@ -2280,7 +2286,8 @@ getUserServerRcvQueueSubs db userId (SMPServer h p kh) onlyNeeded hasService =
       | otherwise = ""
 
 unassocUserServerRcvQueueSubs :: DB.Connection -> UserId -> SMPServer -> IO [RcvQueueSub]
-unassocUserServerRcvQueueSubs db userId (SMPServer h p kh) =
+unassocUserServerRcvQueueSubs db userId srv@(SMPServer h p kh) = do
+  deleteClientService db userId srv
   map toRcvQueueSub
     <$> DB.query
       db
@@ -2295,7 +2302,11 @@ unassocUserServerRcvQueueSubs db userId (SMPServer h p kh) =
       |]
 
 unassocUserServerRcvQueueSubs' :: DB.Connection -> UserId -> SMPServer -> IO ()
-unassocUserServerRcvQueueSubs' db userId (SMPServer h p kh) = DB.execute db removeRcvAssocsQuery (h, p, userId, kh)
+unassocUserServerRcvQueueSubs' db userId srv@(SMPServer h p kh) = do
+  print "111 before deleteClientService"
+  deleteClientService db userId srv
+  print "222 after deleteClientService"
+  DB.execute db removeRcvAssocsQuery (h, p, userId, kh)
 
 unsetQueuesToSubscribe :: DB.Connection -> IO ()
 unsetQueuesToSubscribe db = DB.execute_ db "UPDATE rcv_queues SET to_subscribe = 0 WHERE to_subscribe = 1"

tests/AgentTests/FunctionalAPITests.hs

@@ -479,7 +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
+    fit "should re-subscribe when service ID changed" $ testClientServiceIDChange ps
     it "migrate connections to and from service" $ testMigrateConnectionsToService ps
   describe "Connection switch" $ do
     describe "should switch delivery to the new queue" $
@@ -3722,10 +3722,15 @@ testClientServiceConnection ps = do
 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 <- withSmpServerStoreLogOn ps testPort $ \_ -> runRight $ do
       conns@(sId, uId) <- makeConnection service user
       exchangeGreetings service uId user sId
       pure conns
+    r <- nGet service
+    print r
+    ("", "", SERVICE_DOWN _ (SMP.ServiceSub _ 1 _)) <- pure r
+    ("", "", DOWN _ [_]) <- nGet user
+    pure conns
   _ :: () <- case qs of
     SQSPostgres -> do
 #if defined(dbServerPostgres)
@@ -3738,24 +3743,34 @@ testClientServiceIDChange ps@(_, ASType qs _) = do
       s <- readFile testStoreLogFile
       removeFile testStoreLogFile
       writeFile testStoreLogFile $ unlines $ filter (not . ("NEW_SERVICE" `isPrefixOf`)) $ lines s
+  print 1
   withAgentClientsServers2 (agentCfg, initAgentServersClientService) (agentCfg, initAgentServers) $ \service user -> do
     withSmpServerStoreLogOn ps testPort $ \_ -> runRight $ do
+      liftIO $ print 2
+      liftIO $ threadDelay 250000
       subscribeAllConnections service False Nothing
+      liftIO $ print 3
       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
         ]
+      liftIO $ print 4
       subscribeAllConnections user False Nothing
       ("", "", UP _ [_]) <- nGet user
       exchangeGreetingsMsgId 4 service uId user sId
+      liftIO $ print 5
+    ("", "", SERVICE_DOWN _ (SMP.ServiceSub _ 1 _)) <- nGet service
+    ("", "", DOWN _ [_]) <- nGet user
+    pure ()
   -- 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
+  withAgentClientsServers2 (agentCfg, initAgentServers) (agentCfg, initAgentServers) $ \notService user -> do
     withSmpServerStoreLogOn ps testPort $ \_ -> runRight $ do
+      liftIO $ print 6
       subscribeAllConnections notService False Nothing
+      liftIO $ print 7
       ("", "", UP _ [_]) <- nGet notService
+      liftIO $ print 8
       subscribeAllConnections user False Nothing
       ("", "", UP _ [_]) <- nGet user
       exchangeGreetingsMsgId 6 notService uId user sId
@@ -3856,7 +3871,7 @@ testMigrateConnectionsToService ps = do
       -- the "error" in SERVICE_UP event is expected, because when service was disabled for the user,
       -- the service and associations were not removed, to optimize non-service clients.
       liftIO $ getInAnyOrder service
-        [ \case ("", "", AEvt SAENone (SERVICE_UP _ (SMP.ServiceSubResult (Just (SMP.SSErrorQueueCount 6 0)) (SMP.ServiceSub _ 0 _)))) -> True; _ -> False,
+        [ \case ("", "", AEvt SAENone (SERVICE_UP _ (SMP.ServiceSubResult Nothing (SMP.ServiceSub _ 0 _)))) -> True; _ -> False,
           \case ("", "", AEvt SAENone (SERVICE_ALL _)) -> True; _ -> False
         ]
       service `up` 8

tests/Test.hs

@@ -139,7 +139,7 @@ main = do
             before (pure $ ASType SQSPostgres SMSPostgres) smpProxyTests
 #endif
         -- xdescribe "SMP client agent, server jornal message store" $ agentTests (transport @TLS, ASType SQSMemory SMSJournal)
-        describe "SMP client agent, server memory message store" $ agentTests (transport @TLS, ASType SQSMemory SMSMemory)
+        xdescribe "SMP client agent, server memory message store" $ agentTests (transport @TLS, ASType SQSMemory SMSMemory)
         describe "SMP proxy, jornal message store" $
           before (pure $ ASType SQSMemory SMSJournal) smpProxyTests
         describe "XFTP" $ do