track counts and hashes in smp/ntf servers via triggers, smp server stats for service subscription, update SMP protocol to pass expected count and hash in SSUB/NSSUB commands

Author: epoberezkin

src/Simplex/Messaging/Agent.hs

@@ -200,7 +200,6 @@ import Simplex.Messaging.Protocol
     ErrorType (AUTH),
     MsgBody,
     MsgFlags (..),
-    IdsHash,
     NtfServer,
     ProtoServerWithAuth (..),
     ProtocolServer (..),
@@ -211,6 +210,7 @@ import Simplex.Messaging.Protocol
     SMPMsgMeta,
     SParty (..),
     SProtocolType (..),
+    ServiceSub (..),
     SndPublicAuthKey,
     SubscriptionMode (..),
     UserProtocol,
@@ -466,7 +466,7 @@ resubscribeConnections :: AgentClient -> [ConnId] -> AE (Map ConnId (Either Agen
 resubscribeConnections c = withAgentEnv c . resubscribeConnections' c
 {-# INLINE resubscribeConnections #-}
 
-subscribeClientServices :: AgentClient -> UserId -> AE (Map SMPServer (Either AgentErrorType (Int64, IdsHash)))
+subscribeClientServices :: AgentClient -> UserId -> AE (Map SMPServer (Either AgentErrorType ServiceSub))
 subscribeClientServices c = withAgentEnv c . subscribeClientServices' c
 {-# INLINE subscribeClientServices #-}
 
@@ -1362,15 +1362,15 @@ resubscribeConnections' c connIds = do
   -- union is left-biased, so results returned by subscribeConnections' take precedence
   (`M.union` r) <$> subscribeConnections' c connIds'
 
--- TODO [serts rcv] compare hash with lock
-subscribeClientServices' :: AgentClient -> UserId -> AM (Map SMPServer (Either AgentErrorType (Int64, IdsHash)))
+-- TODO [serts rcv] compare hash. possibly, it should return both expected and returned counts
+subscribeClientServices' :: AgentClient -> UserId -> AM (Map SMPServer (Either AgentErrorType ServiceSub))
 subscribeClientServices' c userId =
   ifM useService subscribe $ throwError $ CMD PROHIBITED "no user service allowed"
   where
     useService = liftIO $ (Just True ==) <$> TM.lookupIO userId (useClientServices c)
     subscribe = do
       srvs <- withStore' c (`getClientServiceServers` userId)
-      lift $ M.fromList . zip srvs <$> mapConcurrently (tryAllErrors' . subscribeClientService c userId) srvs
+      lift $ M.fromList <$> mapConcurrently (\(srv, ServiceSub _ n idsHash) -> fmap (srv,) $ tryAllErrors' $ subscribeClientService c userId srv n idsHash) srvs
 
 -- requesting messages sequentially, to reduce memory usage
 getConnectionMessages' :: AgentClient -> NonEmpty ConnMsgReq -> AM' (NonEmpty (Either AgentErrorType (Maybe SMPMsgMeta)))

src/Simplex/Messaging/Agent/Client.hs

@@ -270,6 +270,7 @@ import Simplex.Messaging.Protocol
     RcvNtfPublicDhKey,
     SMPMsgMeta (..),
     SProtocolType (..),
+    ServiceSub,
     SndPublicAuthKey,
     SubscriptionMode (..),
     NewNtfCreds (..),
@@ -1578,10 +1579,10 @@ subscribeQueues c qs = do
         processSubResults = mapM_ $ uncurry $ processSubResult c sessId
         resubscribe = resubscribeSMPSession c tSess `runReaderT` env
 
-subscribeClientService :: AgentClient -> UserId -> SMPServer -> AM (Int64, IdsHash)
-subscribeClientService c userId srv =
-  withLogClient c NRMBackground (userId, srv, Nothing) B.empty "SUBS" $
-    (`subscribeService` SMP.SRecipientService) . connectedClient
+subscribeClientService :: AgentClient -> UserId -> SMPServer -> Int64 -> IdsHash -> AM ServiceSub
+subscribeClientService c userId srv n idsHash =
+  withLogClient c NRMBackground (userId, srv, Nothing) B.empty "SUBS" $ \(SMPConnectedClient smp _) ->
+    subscribeService smp SMP.SRecipientService n idsHash
 
 activeClientSession :: AgentClient -> SMPTransportSession -> SessionId -> STM Bool
 activeClientSession c tSess sessId = sameSess <$> tryReadSessVar tSess (smpClients c)

src/Simplex/Messaging/Agent/NtfSubSupervisor.hs

@@ -314,7 +314,7 @@ runNtfWorker c srv Worker {doWork} =
               _ -> ((ntfSubConnId sub, INTERNAL "NSACheck - no subscription ID") : errs, subs, subIds)
         updateSub :: DB.Connection -> NtfServer -> UTCTime -> UTCTime -> (NtfSubscription, NtfSubStatus) -> IO (Maybe SMPServer)
         updateSub db ntfServer ts nextCheckTs (sub, status)
-          | ntfShouldSubscribe status =
+          | status `elem` subscribeNtfStatuses =
               let sub' = sub {ntfSubStatus = NASCreated status}
                in Nothing <$ updateNtfSubscription db sub' (NSANtf NSACheck) nextCheckTs
           -- ntf server stopped subscribing to this queue

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

@@ -409,19 +409,20 @@ getClientService db userId srv =
   where
     toService (kh, cert, pk, serviceId_) = ((kh, (cert, pk)), serviceId_)
 
-getClientServiceServers :: DB.Connection -> UserId -> IO [SMPServer]
+getClientServiceServers :: DB.Connection -> UserId -> IO [(SMPServer, ServiceSub)]
 getClientServiceServers db userId =
   map toServer
     <$> DB.query
       db
       [sql|
-        SELECT c.host, c.port, s.key_hash
+        SELECT c.host, c.port, s.key_hash, c.rcv_service_id, c.rcv_service_queue_count, c.rcv_service_queue_ids_hash
         FROM client_services c
         JOIN servers s ON s.host = c.host AND s.port = c.port
       |]
       (Only userId)
   where
-    toServer (host, port, kh) = SMPServer host port kh
+    toServer (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 =

src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20250815_service_certs.hs

@@ -17,7 +17,8 @@ CREATE TABLE client_services(
   service_cert_hash BLOB NOT NULL,
   service_priv_key BLOB NOT NULL,
   rcv_service_id BLOB,
-  rcv_queues_hash BLOB NOT NULL DEFAULT x'',
+  rcv_service_queue_count INTEGER NOT NULL DEFAULT 0,
+  rcv_service_queue_ids_hash BLOB NOT NULL DEFAULT x'00000000000000000000000000000000',
   FOREIGN KEY(host, port) REFERENCES servers ON UPDATE CASCADE ON DELETE RESTRICT
 );
 

src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql

@@ -446,7 +446,8 @@ CREATE TABLE client_services(
   service_cert_hash BLOB NOT NULL,
   service_priv_key BLOB NOT NULL,
   rcv_service_id BLOB,
-  rcv_queues_hash BLOB NOT NULL DEFAULT x'',
+  rcv_service_queue_count INTEGER NOT NULL DEFAULT 0,
+  rcv_service_queue_ids_hash BLOB NOT NULL DEFAULT x'00000000000000000000000000000000',
   FOREIGN KEY(host, port) REFERENCES servers ON UPDATE CASCADE ON DELETE RESTRICT
 );
 CREATE UNIQUE INDEX idx_rcv_queues_ntf ON rcv_queues(host, port, ntf_id);

src/Simplex/Messaging/Client.hs

@@ -892,18 +892,18 @@ nsubResponse_ = \case
 {-# INLINE nsubResponse_ #-}
 
 -- This command is always sent in background request mode
-subscribeService :: forall p. (PartyI p, ServiceParty p) => SMPClient -> SParty p -> ExceptT SMPClientError IO (Int64, IdsHash)
-subscribeService c party = case smpClientService c of
+subscribeService :: forall p. (PartyI p, ServiceParty p) => SMPClient -> SParty p -> Int64 -> IdsHash -> ExceptT SMPClientError IO ServiceSub
+subscribeService c party n idsHash = case smpClientService c of
   Just THClientService {serviceId, serviceKey} -> do
     liftIO $ enablePings c
     sendSMPCommand c NRMBackground (Just (C.APrivateAuthKey C.SEd25519 serviceKey)) serviceId subCmd >>= \case
-      SOKS n idsHash -> pure (n, idsHash)
+      SOKS n' idsHash' -> pure $ ServiceSub serviceId n' idsHash'
       r -> throwE $ unexpectedResponse r
     where
       subCmd :: Command p
       subCmd = case party of
-        SRecipientService -> SUBS
-        SNotifierService -> NSUBS
+        SRecipientService -> SUBS n idsHash
+        SNotifierService -> NSUBS n idsHash
   Nothing -> throwE PCEServiceUnavailable
 
 smpClientService :: SMPClient -> Maybe THClientService

src/Simplex/Messaging/Client/Agent.hs

@@ -45,7 +45,6 @@ import Crypto.Random (ChaChaDRG)
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Constraint (Dict (..))
-import Data.Int (Int64)
 import Data.List.NonEmpty (NonEmpty)
 import qualified Data.List.NonEmpty as L
 import Data.Map.Strict (Map)
@@ -69,10 +68,12 @@ import Simplex.Messaging.Protocol
     ProtocolServer (..),
     QueueId,
     SMPServer,
+    ServiceSub (..),
     SParty (..),
     ServiceParty,
     serviceParty,
-    partyServiceRole
+    partyServiceRole,
+    queueIdsHash,
   )
 import Simplex.Messaging.Session
 import Simplex.Messaging.TMap (TMap)
@@ -91,14 +92,14 @@ data SMPClientAgentEvent
   | CADisconnected SMPServer (NonEmpty QueueId)
   | CASubscribed SMPServer (Maybe ServiceId) (NonEmpty QueueId)
   | CASubError SMPServer (NonEmpty (QueueId, SMPClientError))
-  | CAServiceDisconnected SMPServer (ServiceId, Int64)
-  | CAServiceSubscribed SMPServer (ServiceId, Int64) Int64
-  | CAServiceSubError SMPServer (ServiceId, Int64) SMPClientError
+  | CAServiceDisconnected SMPServer ServiceSub
+  | CAServiceSubscribed {subServer :: SMPServer, expected :: ServiceSub, subscribed :: ServiceSub}
+  | CAServiceSubError SMPServer ServiceSub SMPClientError
   -- CAServiceUnavailable is used when service ID in pending subscription is different from the current service in connection.
   -- This will require resubscribing to all queues associated with this service ID individually, creating new associations.
   -- It may happen if, for example, SMP server deletes service information (e.g. via downgrade and upgrade)
   -- and assigns different service ID to the service certificate.
-  | CAServiceUnavailable SMPServer (ServiceId, Int64)
+  | CAServiceUnavailable SMPServer ServiceSub
 
 data SMPClientAgentConfig = SMPClientAgentConfig
   { smpCfg :: ProtocolClientConfig SMPVersion,
@@ -142,11 +143,11 @@ data SMPClientAgent p = SMPClientAgent
     -- Only one service subscription can exist per server with this agent.
     -- With correctly functioning SMP server, queue and service subscriptions can't be
     -- active at the same time.
-    activeServiceSubs :: TMap SMPServer (TVar (Maybe ((ServiceId, Int64), SessionId))),
+    activeServiceSubs :: TMap SMPServer (TVar (Maybe (ServiceSub, SessionId))),
     activeQueueSubs :: TMap SMPServer (TMap QueueId (SessionId, C.APrivateAuthKey)),
     -- Pending service subscriptions can co-exist with pending queue subscriptions
     -- on the same SMP server during subscriptions being transitioned from per-queue to service.
-    pendingServiceSubs :: TMap SMPServer (TVar (Maybe (ServiceId, Int64))),
+    pendingServiceSubs :: TMap SMPServer (TVar (Maybe ServiceSub)),
     pendingQueueSubs :: TMap SMPServer (TMap QueueId C.APrivateAuthKey),
     smpSubWorkers :: TMap SMPServer (SessionVar (Async ())),
     workerSeq :: TVar Int
@@ -256,7 +257,7 @@ connectClient ca@SMPClientAgent {agentCfg, smpClients, smpSessions, msgQ, random
       removeClientAndSubs smp >>= serverDown
       logInfo . decodeUtf8 $ "Agent disconnected from " <> showServer srv
 
-    removeClientAndSubs :: SMPClient -> IO (Maybe (ServiceId, Int64), Maybe (Map QueueId C.APrivateAuthKey))
+    removeClientAndSubs :: SMPClient -> IO (Maybe ServiceSub, Maybe (Map QueueId C.APrivateAuthKey))
     removeClientAndSubs smp = do
       -- Looking up subscription vars outside of STM transaction to reduce re-evaluation.
       -- It is possible because these vars are never removed, they are only added.
@@ -287,7 +288,7 @@ connectClient ca@SMPClientAgent {agentCfg, smpClients, smpSessions, msgQ, random
             then pure Nothing
             else Just subs <$ addSubs_ (pendingQueueSubs ca) srv subs
 
-    serverDown :: (Maybe (ServiceId, Int64), Maybe (Map QueueId C.APrivateAuthKey)) -> IO ()
+    serverDown :: (Maybe ServiceSub, Maybe (Map QueueId C.APrivateAuthKey)) -> IO ()
     serverDown (sSub, qSubs) = do
       mapM_ (notify ca . CAServiceDisconnected srv) sSub
       let qIds = L.nonEmpty . M.keys =<< qSubs
@@ -317,7 +318,7 @@ reconnectClient ca@SMPClientAgent {active, agentCfg, smpSubWorkers, workerSeq} s
           loop
     ProtocolClientConfig {networkConfig = NetworkConfig {tcpConnectTimeout}} = smpCfg agentCfg
     noPending (sSub, qSubs) = isNothing sSub && maybe True M.null qSubs
-    getPending :: Monad m => (forall a. SMPServer -> TMap SMPServer a -> m (Maybe a)) -> (forall a. TVar a -> m a) -> m (Maybe (ServiceId, Int64), Maybe (Map QueueId C.APrivateAuthKey))
+    getPending :: Monad m => (forall a. SMPServer -> TMap SMPServer a -> m (Maybe a)) -> (forall a. TVar a -> m a) -> m (Maybe ServiceSub, Maybe (Map QueueId C.APrivateAuthKey))
     getPending lkup rd = do
       sSub <- lkup srv (pendingServiceSubs ca) $>>= rd
       qSubs <- lkup srv (pendingQueueSubs ca) >>= mapM rd
@@ -329,7 +330,7 @@ reconnectClient ca@SMPClientAgent {active, agentCfg, smpSubWorkers, workerSeq} s
       whenM (isEmptyTMVar $ sessionVar v) retry
       removeSessVar v srv smpSubWorkers
 
-reconnectSMPClient :: forall p. SMPClientAgent p -> SMPServer -> (Maybe (ServiceId, Int64), Maybe (Map QueueId C.APrivateAuthKey)) -> ExceptT SMPClientError IO ()
+reconnectSMPClient :: forall p. SMPClientAgent p -> SMPServer -> (Maybe ServiceSub, Maybe (Map QueueId C.APrivateAuthKey)) -> ExceptT SMPClientError IO ()
 reconnectSMPClient ca@SMPClientAgent {agentCfg, agentParty} srv (sSub_, qSubs_) =
   withSMP ca srv $ \smp -> liftIO $ case serviceParty agentParty of
     Just Dict -> resubscribe smp
@@ -430,7 +431,7 @@ smpSubscribeQueues ca smp srv subs = do
       let acc@(_, _, (qOks, sQs), notPending) = foldr (groupSub pending) (False, [], ([], []), []) (L.zip subs rs)
       unless (null qOks) $ addActiveSubs ca srv qOks
       unless (null sQs) $ forM_ smpServiceId $ \serviceId ->
-        updateActiveServiceSub ca srv ((serviceId, fromIntegral $ length sQs), sessId)
+        updateActiveServiceSub ca srv (ServiceSub serviceId (fromIntegral $ length sQs) (queueIdsHash sQs), sessId)
       unless (null notPending) $ removePendingSubs ca srv notPending
       pure acc
     sessId = sessionId $ thParams smp
@@ -454,40 +455,40 @@ smpSubscribeQueues ca smp srv subs = do
     notify_ :: (SMPServer -> NonEmpty a -> SMPClientAgentEvent) -> [a] -> IO ()
     notify_ evt qs = mapM_ (notify ca . evt srv) $ L.nonEmpty qs
 
-subscribeServiceNtfs :: SMPClientAgent 'NotifierService -> SMPServer -> (ServiceId, Int64) -> IO ()
+subscribeServiceNtfs :: SMPClientAgent 'NotifierService -> SMPServer -> ServiceSub -> IO ()
 subscribeServiceNtfs = subscribeService_
 {-# INLINE subscribeServiceNtfs #-}
 
-subscribeService_ :: (PartyI p, ServiceParty p) => SMPClientAgent p -> SMPServer -> (ServiceId, Int64) -> IO ()
+subscribeService_ :: (PartyI p, ServiceParty p) => SMPClientAgent p -> SMPServer -> ServiceSub -> IO ()
 subscribeService_ ca srv serviceSub = do
   atomically $ setPendingServiceSub ca srv $ Just serviceSub
   runExceptT (getSMPServerClient' ca srv) >>= \case
     Right smp -> smpSubscribeService ca smp srv serviceSub
     Left _ -> pure () -- no call to reconnectClient - failing getSMPServerClient' does that
 
-smpSubscribeService :: (PartyI p, ServiceParty p) => SMPClientAgent p -> SMPClient -> SMPServer -> (ServiceId, Int64) -> IO ()
-smpSubscribeService ca smp srv serviceSub@(serviceId, _) = case smpClientService smp of
+smpSubscribeService :: (PartyI p, ServiceParty p) => SMPClientAgent p -> SMPClient -> SMPServer -> ServiceSub -> IO ()
+smpSubscribeService ca smp srv serviceSub@(ServiceSub serviceId n idsHash) = case smpClientService smp of
   Just service | serviceAvailable service -> subscribe
   _ -> notifyUnavailable
   where
     subscribe = do
-      r <- runExceptT $ subscribeService smp $ agentParty ca
+      r <- runExceptT $ subscribeService smp (agentParty ca) n idsHash
       ok <-
         atomically $
           ifM
             (activeClientSession ca smp srv)
             (True <$ processSubscription r)
             (pure False)
       if ok
-        then case r of -- TODO [certs rcv] compare hash
-          Right (n, _idsHash) -> notify ca $ CAServiceSubscribed srv serviceSub n
+        then case r of
+          Right serviceSub' -> notify ca $ CAServiceSubscribed srv serviceSub serviceSub'
           Left e
             | smpClientServiceError e -> notifyUnavailable
             | temporaryClientError e -> reconnectClient ca srv
             | otherwise -> notify ca $ CAServiceSubError srv serviceSub e
         else reconnectClient ca srv
-    processSubscription = mapM_ $ \(n, _idsHash) -> do -- TODO [certs rcv] validate hash here?
-      setActiveServiceSub ca srv $ Just ((serviceId, n), sessId)
+    processSubscription = mapM_ $ \serviceSub' -> do -- TODO [certs rcv] validate hash here?
+      setActiveServiceSub ca srv $ Just (serviceSub', sessId)
       setPendingServiceSub ca srv Nothing
     serviceAvailable THClientService {serviceRole, serviceId = serviceId'} =
       serviceId == serviceId' && partyServiceRole (agentParty ca) == serviceRole
@@ -529,11 +530,11 @@ addSubs_ subs srv ss =
     Just m -> TM.union ss m
     _ -> TM.insertM srv (newTVar ss) subs
 
-setActiveServiceSub :: SMPClientAgent p -> SMPServer -> Maybe ((ServiceId, Int64), SessionId) -> STM ()
+setActiveServiceSub :: SMPClientAgent p -> SMPServer -> Maybe (ServiceSub, SessionId) -> STM ()
 setActiveServiceSub = setServiceSub_ activeServiceSubs
 {-# INLINE setActiveServiceSub #-}
 
-setPendingServiceSub :: SMPClientAgent p -> SMPServer -> Maybe (ServiceId, Int64) -> STM ()
+setPendingServiceSub :: SMPClientAgent p -> SMPServer -> Maybe ServiceSub -> STM ()
 setPendingServiceSub = setServiceSub_ pendingServiceSubs
 {-# INLINE setPendingServiceSub #-}
 
@@ -548,12 +549,12 @@ setServiceSub_ subsSel ca srv sub =
     Just v -> writeTVar v sub
     Nothing -> TM.insertM srv (newTVar sub) (subsSel ca)
 
-updateActiveServiceSub :: SMPClientAgent p -> SMPServer -> ((ServiceId, Int64), SessionId) -> STM ()
-updateActiveServiceSub ca srv sub@((serviceId', n'), sessId') =
+updateActiveServiceSub :: SMPClientAgent p -> SMPServer -> (ServiceSub, SessionId) -> STM ()
+updateActiveServiceSub ca srv sub@(ServiceSub serviceId' n' idsHash', sessId') =
   TM.lookup srv (activeServiceSubs ca) >>= \case
     Just v -> modifyTVar' v $ \case
-      Just ((serviceId, n), sessId) | serviceId == serviceId' && sessId == sessId' ->
-        Just ((serviceId, n + n'), sessId)
+      Just (ServiceSub serviceId n idsHash, sessId) | serviceId == serviceId' && sessId == sessId' ->
+        Just (ServiceSub serviceId (n + n') (idsHash <> idsHash'), sessId)
       _ -> Just sub
     Nothing -> TM.insertM srv (newTVar $ Just sub) (activeServiceSubs ca)
 

src/Simplex/Messaging/Crypto.hs

@@ -177,6 +177,7 @@ module Simplex.Messaging.Crypto
     sha512Hash,
     sha3_256,
     sha3_384,
+    md5Hash,
 
     -- * Message padding / un-padding
     canPad,
@@ -215,7 +216,7 @@ import Crypto.Cipher.AES (AES256)
 import qualified Crypto.Cipher.Types as AES
 import qualified Crypto.Cipher.XSalsa as XSalsa
 import qualified Crypto.Error as CE
-import Crypto.Hash (Digest, SHA3_256, SHA3_384, SHA256 (..), SHA512 (..), hash, hashDigestSize)
+import Crypto.Hash (Digest, MD5, SHA3_256, SHA3_384, SHA256 (..), SHA512 (..), hash, hashDigestSize)
 import qualified Crypto.KDF.HKDF as H
 import qualified Crypto.MAC.Poly1305 as Poly1305
 import qualified Crypto.PubKey.Curve25519 as X25519
@@ -1006,6 +1007,9 @@ sha3_384 :: ByteString -> ByteString
 sha3_384 = BA.convert . (hash :: ByteString -> Digest SHA3_384)
 {-# INLINE sha3_384 #-}
 
+md5Hash :: ByteString -> ByteString
+md5Hash = BA.convert . (hash :: ByteString -> Digest MD5)
+
 -- | AEAD-GCM encryption with associated data.
 --
 -- Used as part of double ratchet encryption.

src/Simplex/Messaging/Notifications/Protocol.hs

@@ -489,17 +489,9 @@ data NtfSubStatus
     NSErr ByteString
   deriving (Eq, Ord, Show)
 
-ntfShouldSubscribe :: NtfSubStatus -> Bool
-ntfShouldSubscribe = \case
-  NSNew -> True
-  NSPending -> True
-  NSActive -> True
-  NSInactive -> True
-  NSEnd -> False
-  NSDeleted -> False
-  NSAuth -> False
-  NSService -> True
-  NSErr _ -> False
+-- if these statuses change, the queue ID hashes for services need to be updated in a new migration (see m20250830_queue_ids_hash)
+subscribeNtfStatuses :: [NtfSubStatus]
+subscribeNtfStatuses = [NSNew, NSPending, NSActive, NSInactive]
 
 instance Encoding NtfSubStatus where
   smpEncode = \case