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

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,24 +455,24 @@ 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
@@ -479,15 +480,15 @@ smpSubscribeService ca smp srv serviceSub@(serviceId, _) = case smpClientService
             (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

src/Simplex/Messaging/Notifications/Server.hs

@@ -62,7 +62,7 @@ import Simplex.Messaging.Notifications.Server.Store (NtfSTMStore, TokenNtfMessag
 import Simplex.Messaging.Notifications.Server.Store.Postgres
 import Simplex.Messaging.Notifications.Server.Store.Types
 import Simplex.Messaging.Notifications.Transport
-import Simplex.Messaging.Protocol (EntityId (..), ErrorType (..), NotifierId, Party (..), ProtocolServer (host), SMPServer, ServiceId, SignedTransmission, Transmission, pattern NoEntity, pattern SMPServer, encodeTransmission, tGetServer, tPut)
+import Simplex.Messaging.Protocol (EntityId (..), ErrorType (..), NotifierId, Party (..), ProtocolServer (host), SMPServer, ServiceSub (..), SignedTransmission, Transmission, pattern NoEntity, pattern SMPServer, encodeTransmission, tGetServer, tPut)
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Server
 import Simplex.Messaging.Server.Control (CPClientRole (..))
@@ -257,9 +257,9 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg, startOptions}
       srvSubscribers <- getSMPWorkerMetrics a smpSubscribers
       srvClients <- getSMPWorkerMetrics a smpClients
       srvSubWorkers <- getSMPWorkerMetrics a smpSubWorkers
-      ntfActiveServiceSubs <- getSMPServiceSubMetrics a activeServiceSubs $ snd . fst
+      ntfActiveServiceSubs <- getSMPServiceSubMetrics a activeServiceSubs $ smpQueueCount . fst
       ntfActiveQueueSubs <- getSMPSubMetrics a activeQueueSubs
-      ntfPendingServiceSubs <- getSMPServiceSubMetrics a pendingServiceSubs snd
+      ntfPendingServiceSubs <- getSMPServiceSubMetrics a pendingServiceSubs smpQueueCount
       ntfPendingQueueSubs <- getSMPSubMetrics a pendingQueueSubs
       smpSessionCount <- M.size <$> readTVarIO smpSessions
       apnsPushQLength <- atomically $ lengthTBQueue pushQ
@@ -452,13 +452,13 @@ resubscribe NtfSubscriber {smpAgent = ca} = do
     counts <- mapConcurrently (subscribeSrvSubs ca st batchSize) srvs
     logNote $ "Completed all SMP resubscriptions for " <> tshow (length srvs) <> " servers (" <> tshow (sum counts) <> " subscriptions)"
 
-subscribeSrvSubs :: SMPClientAgent 'NotifierService -> NtfPostgresStore -> Int -> (SMPServer, Int64, Maybe (ServiceId, Int64)) -> IO Int
+subscribeSrvSubs :: SMPClientAgent 'NotifierService -> NtfPostgresStore -> Int -> (SMPServer, Int64, Maybe ServiceSub) -> IO Int
 subscribeSrvSubs ca st batchSize (srv, srvId, service_) = do
   let srvStr = safeDecodeUtf8 (strEncode $ L.head $ host srv)
   logNote $ "Starting SMP resubscriptions for " <> srvStr
-  forM_ service_ $ \(serviceId, n) -> do
-    logNote $ "Subscribing service to " <> srvStr <> " with " <> tshow n <> " associated queues"
-    subscribeServiceNtfs ca srv (serviceId, n)
+  forM_ service_ $ \serviceSub -> do
+    logNote $ "Subscribing service to " <> srvStr <> " with " <> tshow (smpQueueCount serviceSub) <> " associated queues"
+    subscribeServiceNtfs ca srv serviceSub
   n <- subscribeLoop 0 Nothing
   logNote $ "Completed SMP resubscriptions for " <> srvStr <> " (" <> tshow n <> " subscriptions)"
   pure n
@@ -576,7 +576,7 @@ ntfSubscriber NtfSubscriber {smpAgent = ca@SMPClientAgent {msgQ, agentQ}} =
               -- TODO [certs] resubscribe queues with statuses NSErr and NSService
           CAServiceDisconnected srv serviceSub ->
             logNote $ "SMP server service disconnected " <> showService srv serviceSub
-          CAServiceSubscribed srv serviceSub@(_, expected) n
+          CAServiceSubscribed srv serviceSub@(ServiceSub _ expected _) (ServiceSub _ n _) -- TODO [certs rcv] compare hash
             | expected == n -> logNote msg
             | otherwise -> logWarn $ msg <> ", confirmed subs: " <> tshow n
             where
@@ -593,7 +593,8 @@ ntfSubscriber NtfSubscriber {smpAgent = ca@SMPClientAgent {msgQ, agentQ}} =
                 void $ subscribeSrvSubs ca st batchSize (srv, srvId, Nothing)
               Left e -> logError $ "SMP server update and resubscription error " <> tshow e
       where
-        showService srv (serviceId, n) = showServer' srv  <> ", service ID " <> decodeLatin1 (strEncode serviceId) <> ", " <> tshow n <> " subs"
+        -- TODO [certs rcv] compare hash
+        showService srv (ServiceSub serviceId n _idsHash) = showServer' srv  <> ", service ID " <> decodeLatin1 (strEncode serviceId) <> ", " <> tshow n <> " subs"
 
     logSubErrors :: SMPServer -> NonEmpty (SMP.NotifierId, NtfSubStatus) -> Int -> IO ()
     logSubErrors srv subs updated = forM_ (L.group $ L.sort $ L.map snd subs) $ \ss -> do

src/Simplex/Messaging/Notifications/Server/Stats.hs

@@ -17,6 +17,7 @@ import Simplex.Messaging.Server.Stats
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 
+-- TODO [certs rcv] track service subscriptions and count/hash diffs for own and other servers + prometheus
 data NtfServerStats = NtfServerStats
   { fromTime :: IORef UTCTime,
     tknCreated :: IORef Int,

src/Simplex/Messaging/Notifications/Server/Store/Migrations.hs

@@ -112,27 +112,48 @@ m20250830_queue_ids_hash =
   createXorHashFuncs
     <> T.pack
       [r|
-ALTER TABLE smp_servers ADD COLUMN smp_notifier_ids_hash BYTEA NOT NULL DEFAULT '\x00000000000000000000000000000000';
+ALTER TABLE smp_servers
+  ADD COLUMN smp_notifier_count BIGINT NOT NULL DEFAULT 0,
+  ADD COLUMN smp_notifier_ids_hash BYTEA NOT NULL DEFAULT '\x00000000000000000000000000000000';
 
-WITH hashes AS (
-  SELECT
-    s.smp_server_id,
-    xor_aggregate(public.digest(s.smp_notifier_id, 'md5')) AS notifier_hash
-  FROM subscriptions s
-  WHERE s.ntf_service_assoc = true
-  GROUP BY s.smp_server_id
-)
-UPDATE smp_servers srv
-SET smp_notifier_ids_hash = COALESCE(h.notifier_hash, '\x00000000000000000000000000000000')
-FROM hashes h
-WHERE srv.smp_server_id = h.smp_server_id;
+CREATE OR REPLACE FUNCTION should_subscribe_status(p_status TEXT) RETURNS BOOLEAN
+LANGUAGE plpgsql IMMUTABLE STRICT
+AS $$
+BEGIN
+  RETURN p_status IN ('NEW', 'PENDING', 'ACTIVE', 'INACTIVE');
+END;
+$$;
 
-CREATE OR REPLACE FUNCTION update_ids_hash(p_server_id BIGINT, p_notifier_id BYTEA) RETURNS void
+CREATE OR REPLACE FUNCTION update_all_aggregates() RETURNS void
+LANGUAGE plpgsql
+AS $$
+BEGIN
+  WITH acc AS (
+    SELECT
+      s.smp_server_id,
+      count(smp_notifier_id) as notifier_count,
+      xor_aggregate(public.digest(s.smp_notifier_id, 'md5')) AS notifier_hash
+    FROM subscriptions s
+    WHERE s.ntf_service_assoc = true AND should_subscribe_status(s.status)
+    GROUP BY s.smp_server_id
+  )
+  UPDATE smp_servers srv
+  SET smp_notifier_count = COALESCE(acc.notifier_count, 0),
+      smp_notifier_ids_hash = COALESCE(acc.notifier_hash, '\x00000000000000000000000000000000')
+  FROM acc
+  WHERE srv.smp_server_id = acc.smp_server_id;
+END;
+$$;
+
+SELECT update_all_aggregates();
+
+CREATE OR REPLACE FUNCTION update_aggregates(p_server_id BIGINT, p_change BIGINT, p_notifier_id BYTEA) RETURNS void
 LANGUAGE plpgsql
 AS $$
 BEGIN
   UPDATE smp_servers
-  SET smp_notifier_ids_hash = xor_combine(smp_notifier_ids_hash, public.digest(p_notifier_id, 'md5'))
+  SET smp_notifier_count = smp_notifier_count + p_change,
+      smp_notifier_ids_hash = xor_combine(smp_notifier_ids_hash, public.digest(p_notifier_id, 'md5'))
   WHERE smp_server_id = p_server_id;
 END;
 $$;
@@ -141,8 +162,8 @@ CREATE OR REPLACE FUNCTION on_subscription_insert() RETURNS TRIGGER
 LANGUAGE plpgsql
 AS $$
 BEGIN
-  IF NEW.ntf_service_assoc = true THEN
-    PERFORM update_ids_hash(NEW.smp_server_id, NEW.smp_notifier_id);
+  IF NEW.ntf_service_assoc = true AND should_subscribe_status(NEW.status) THEN
+    PERFORM update_aggregates(NEW.smp_server_id, 1, NEW.smp_notifier_id);
   END IF;
   RETURN NEW;
 END;
@@ -152,8 +173,8 @@ CREATE OR REPLACE FUNCTION on_subscription_delete() RETURNS TRIGGER
 LANGUAGE plpgsql
 AS $$
 BEGIN
-  IF OLD.ntf_service_assoc = true THEN
-    PERFORM update_ids_hash(OLD.smp_server_id, OLD.smp_notifier_id);
+  IF OLD.ntf_service_assoc = true AND should_subscribe_status(OLD.status) THEN
+    PERFORM update_aggregates(OLD.smp_server_id, -1, OLD.smp_notifier_id);
   END IF;
   RETURN OLD;
 END;
@@ -163,8 +184,12 @@ CREATE OR REPLACE FUNCTION on_subscription_update() RETURNS TRIGGER
 LANGUAGE plpgsql
 AS $$
 BEGIN
-  IF OLD.ntf_service_assoc != NEW.ntf_service_assoc THEN
-    PERFORM update_ids_hash(NEW.smp_server_id, NEW.smp_notifier_id);
+  IF OLD.ntf_service_assoc = true AND should_subscribe_status(OLD.status) THEN
+    IF NOT (NEW.ntf_service_assoc = true AND should_subscribe_status(NEW.status)) THEN
+      PERFORM update_aggregates(OLD.smp_server_id, -1, OLD.smp_notifier_id);
+    END IF;
+  ELSIF NEW.ntf_service_assoc = true AND should_subscribe_status(NEW.status) THEN
+    PERFORM update_aggregates(NEW.smp_server_id, 1, NEW.smp_notifier_id);
   END IF;
   RETURN NEW;
 END;
@@ -199,8 +224,13 @@ DROP FUNCTION on_subscription_insert;
 DROP FUNCTION on_subscription_delete;
 DROP FUNCTION on_subscription_update;
 
-DROP FUNCTION update_ids_hash;
+DROP FUNCTION update_aggregates;
+DROP FUNCTION update_all_aggregates;
 
-ALTER TABLE smp_servers DROP COLUMN smp_notifier_ids_hash;
+DROP FUNCTION should_subscribe_status;
+
+ALTER TABLE smp_servers
+  DROP COLUMN smp_notifier_count,
+  DROP COLUMN smp_notifier_ids_hash;
     |]
     <> dropXorHashFuncs

src/Simplex/Messaging/Notifications/Server/Store/Postgres.hs

@@ -64,7 +64,7 @@ import Simplex.Messaging.Notifications.Server.Store.Migrations
 import Simplex.Messaging.Notifications.Server.Store.Types
 import Simplex.Messaging.Notifications.Server.StoreLog
 import Simplex.Messaging.Parsers (parseAll)
-import Simplex.Messaging.Protocol (EntityId (..), EncNMsgMeta, ErrorType (..), NotifierId, NtfPrivateAuthKey, NtfPublicAuthKey, SMPServer, ServiceId, pattern SMPServer)
+import Simplex.Messaging.Protocol (EntityId (..), EncNMsgMeta, ErrorType (..), IdsHash (..), NotifierId, NtfPrivateAuthKey, NtfPublicAuthKey, SMPServer, ServiceId, ServiceSub (..), pattern SMPServer)
 import Simplex.Messaging.Server.QueueStore (RoundedSystemTime, getSystemDate)
 import Simplex.Messaging.Server.QueueStore.Postgres (handleDuplicate, withLog_)
 import Simplex.Messaging.Server.QueueStore.Postgres.Config (PostgresStoreCfg (..))
@@ -237,7 +237,7 @@ updateTknCronInterval st tknId cronInt =
 
 -- Reads servers that have subscriptions that need subscribing.
 -- It is executed on server start, and it is supposed to crash on database error
-getUsedSMPServers :: NtfPostgresStore -> IO [(SMPServer, Int64, Maybe (ServiceId, Int64))]
+getUsedSMPServers :: NtfPostgresStore -> IO [(SMPServer, Int64, Maybe ServiceSub)]
 getUsedSMPServers st =
   withTransaction (dbStore st) $ \db ->
     map rowToSrvSubs <$>
@@ -245,25 +245,17 @@ getUsedSMPServers st =
         db
         [sql|
           SELECT
-            p.smp_host, p.smp_port, p.smp_keyhash, p.smp_server_id, p.ntf_service_id,
-            SUM(CASE WHEN s.ntf_service_assoc THEN s.subs_count ELSE 0 END) :: BIGINT as service_subs_count
-          FROM smp_servers p
-          JOIN (
-            SELECT
-              smp_server_id,
-              ntf_service_assoc,
-              COUNT(1) as subs_count
-            FROM subscriptions
-            WHERE status IN ?
-            GROUP BY smp_server_id, ntf_service_assoc
-          ) s ON s.smp_server_id = p.smp_server_id
-          GROUP BY p.smp_host, p.smp_port, p.smp_keyhash, p.smp_server_id, p.ntf_service_id
+            smp_host, smp_port, smp_keyhash, smp_server_id,
+            ntf_service_id, smp_notifier_count, smp_notifier_ids_hash
+          FROM smp_servers
+          WHERE EXISTS (SELECT 1 FROM subscriptions WHERE status IN ?)
         |]
-        (Only (In [NSNew, NSPending, NSActive, NSInactive]))
+        (Only (In subscribeNtfStatuses))
   where
-    rowToSrvSubs :: SMPServerRow :. (Int64, Maybe ServiceId, Int64) -> (SMPServer, Int64, Maybe (ServiceId, Int64))
-    rowToSrvSubs ((host, port, kh) :. (srvId, serviceId_, subsCount)) =
-      (SMPServer host port kh, srvId, (,subsCount) <$> serviceId_)
+    rowToSrvSubs :: SMPServerRow :. (Int64, Maybe ServiceId, Int64, IdsHash) -> (SMPServer, Int64, Maybe ServiceSub)
+    rowToSrvSubs ((host, port, kh) :. (srvId, serviceId_, n, idsHash)) =
+      let service_ = (\serviceId -> ServiceSub serviceId n idsHash) <$> serviceId_
+       in (SMPServer host port kh, srvId, service_)
 
 getServerNtfSubscriptions :: NtfPostgresStore -> Int64 -> Maybe NtfSubscriptionId -> Int -> IO (Either ErrorType [ServerNtfSub])
 getServerNtfSubscriptions st srvId afterSubId_ count =
@@ -271,9 +263,9 @@ getServerNtfSubscriptions st srvId afterSubId_ count =
     subs <-
       map toServerNtfSub <$> case afterSubId_ of
         Nothing ->
-          DB.query db (query <> orderLimit) (srvId, statusIn, count)
+          DB.query db (query <> orderLimit) (srvId, In subscribeNtfStatuses, count)
         Just afterSubId ->
-          DB.query db (query <> " AND subscription_id > ?" <> orderLimit) (srvId, statusIn, afterSubId, count)
+          DB.query db (query <> " AND subscription_id > ?" <> orderLimit) (srvId, In subscribeNtfStatuses, afterSubId, count)
     void $
       DB.executeMany
         db
@@ -294,7 +286,6 @@ getServerNtfSubscriptions st srvId afterSubId_ count =
         WHERE smp_server_id = ? AND NOT ntf_service_assoc AND status IN ?
       |]
     orderLimit = " ORDER BY subscription_id LIMIT ?"
-    statusIn = In [NSNew, NSPending, NSActive, NSInactive]
     toServerNtfSub (ntfSubId, notifierId, notifierKey) = (ntfSubId, (notifierId, notifierKey))
 
 -- Returns token and subscription.

src/Simplex/Messaging/Notifications/Server/Store/ntf_server_schema.sql

@@ -20,8 +20,8 @@ CREATE FUNCTION ntf_server.on_subscription_delete() RETURNS trigger
     LANGUAGE plpgsql
     AS $$
 BEGIN
-  IF OLD.ntf_service_assoc = true THEN
-    PERFORM update_ids_hash(OLD.smp_server_id, OLD.smp_notifier_id);
+  IF OLD.ntf_service_assoc = true AND should_subscribe_status(OLD.status) THEN
+    PERFORM update_aggregates(OLD.smp_server_id, -1, OLD.smp_notifier_id);
   END IF;
   RETURN OLD;
 END;
@@ -33,8 +33,8 @@ CREATE FUNCTION ntf_server.on_subscription_insert() RETURNS trigger
     LANGUAGE plpgsql
     AS $$
 BEGIN
-  IF NEW.ntf_service_assoc = true THEN
-    PERFORM update_ids_hash(NEW.smp_server_id, NEW.smp_notifier_id);
+  IF NEW.ntf_service_assoc = true AND should_subscribe_status(NEW.status) THEN
+    PERFORM update_aggregates(NEW.smp_server_id, 1, NEW.smp_notifier_id);
   END IF;
   RETURN NEW;
 END;
@@ -46,8 +46,12 @@ CREATE FUNCTION ntf_server.on_subscription_update() RETURNS trigger
     LANGUAGE plpgsql
     AS $$
 BEGIN
-  IF OLD.ntf_service_assoc != NEW.ntf_service_assoc THEN
-    PERFORM update_ids_hash(NEW.smp_server_id, NEW.smp_notifier_id);
+  IF OLD.ntf_service_assoc = true AND should_subscribe_status(OLD.status) THEN
+    IF NOT (NEW.ntf_service_assoc = true AND should_subscribe_status(NEW.status)) THEN
+      PERFORM update_aggregates(OLD.smp_server_id, -1, OLD.smp_notifier_id);
+    END IF;
+  ELSIF NEW.ntf_service_assoc = true AND should_subscribe_status(NEW.status) THEN
+    PERFORM update_aggregates(NEW.smp_server_id, 1, NEW.smp_notifier_id);
   END IF;
   RETURN NEW;
 END;
@@ -55,18 +59,52 @@ $$;
 
 
 
-CREATE FUNCTION ntf_server.update_ids_hash(p_server_id bigint, p_notifier_id bytea) RETURNS void
+CREATE FUNCTION ntf_server.should_subscribe_status(p_status text) RETURNS boolean
+    LANGUAGE plpgsql IMMUTABLE STRICT
+    AS $$
+BEGIN
+  RETURN p_status IN ('NEW', 'PENDING', 'ACTIVE', 'INACTIVE');
+END;
+$$;
+
+
+
+CREATE FUNCTION ntf_server.update_aggregates(p_server_id bigint, p_change bigint, p_notifier_id bytea) RETURNS void
     LANGUAGE plpgsql
     AS $$
 BEGIN
   UPDATE smp_servers
-  SET smp_notifier_ids_hash = xor_combine(smp_notifier_ids_hash, public.digest(p_notifier_id, 'md5'))
+  SET smp_notifier_count = smp_notifier_count + p_change,
+      smp_notifier_ids_hash = xor_combine(smp_notifier_ids_hash, public.digest(p_notifier_id, 'md5'))
   WHERE smp_server_id = p_server_id;
 END;
 $$;
 
 
 
+CREATE FUNCTION ntf_server.update_all_aggregates() RETURNS void
+    LANGUAGE plpgsql
+    AS $$
+BEGIN
+  WITH acc AS (
+    SELECT
+      s.smp_server_id,
+      count(smp_notifier_id) as notifier_count,
+      xor_aggregate(public.digest(s.smp_notifier_id, 'md5')) AS notifier_hash
+    FROM subscriptions s
+    WHERE s.ntf_service_assoc = true AND should_subscribe_status(s.status)
+    GROUP BY s.smp_server_id
+  )
+  UPDATE smp_servers srv
+  SET smp_notifier_count = COALESCE(acc.notifier_count, 0),
+      smp_notifier_ids_hash = COALESCE(acc.notifier_hash, '\x00000000000000000000000000000000')
+  FROM acc
+  WHERE srv.smp_server_id = acc.smp_server_id;
+END;
+$$;
+
+
+
 CREATE FUNCTION ntf_server.xor_combine(state bytea, value bytea) RETURNS bytea
     LANGUAGE plpgsql IMMUTABLE STRICT
     AS $$
@@ -133,6 +171,7 @@ CREATE TABLE ntf_server.smp_servers (
     smp_port text NOT NULL,
     smp_keyhash bytea NOT NULL,
     ntf_service_id bytea,
+    smp_notifier_count bigint DEFAULT 0 NOT NULL,
     smp_notifier_ids_hash bytea DEFAULT '\x00000000000000000000000000000000'::bytea NOT NULL
 );
 

src/Simplex/Messaging/Protocol.hs

@@ -139,7 +139,10 @@ module Simplex.Messaging.Protocol
     RcvMessage (..),
     MsgId,
     MsgBody,
-    IdsHash,
+    IdsHash (..),
+    ServiceSub (..),
+    queueIdsHash,
+    queueIdHash,
     MaxMessageLen,
     MaxRcvMessageLen,
     EncRcvMsgBody (..),
@@ -220,6 +223,8 @@ import qualified Data.Aeson.TH as J
 import Data.Attoparsec.ByteString.Char8 (Parser, (<?>))
 import qualified Data.Attoparsec.ByteString.Char8 as A
 import Data.Bifunctor (bimap, first)
+import Data.Bits (xor)
+import qualified Data.ByteString as BS
 import qualified Data.ByteString.Base64 as B64
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
@@ -228,6 +233,7 @@ import Data.Constraint (Dict (..))
 import Data.Functor (($>))
 import Data.Int (Int64)
 import Data.Kind
+import Data.List (foldl')
 import Data.List.NonEmpty (NonEmpty (..))
 import qualified Data.List.NonEmpty as L
 import Data.Maybe (isJust, isNothing)
@@ -237,7 +243,7 @@ import qualified Data.Text as T
 import Data.Text.Encoding (decodeLatin1, encodeUtf8)
 import Data.Time.Clock.System (SystemTime (..), systemToUTCTime)
 import Data.Type.Equality
-import Data.Word (Word16)
+import Data.Word (Word8, Word16)
 import GHC.TypeLits (ErrorMessage (..), TypeError, type (+))
 import qualified GHC.TypeLits as TE
 import qualified GHC.TypeLits as Type
@@ -542,7 +548,8 @@ data Command (p :: Party) where
   NEW :: NewQueueReq -> Command Creator
   SUB :: Command Recipient
   -- | subscribe all associated queues. Service ID must be used as entity ID, and service session key must sign the command.
-  SUBS :: Command RecipientService
+  -- Parameters are expected queue count and hash of all subscribed queues, it allows to monitor "state drift" on the server
+  SUBS :: Int64 -> IdsHash -> Command RecipientService
   KEY :: SndPublicAuthKey -> Command Recipient
   RKEY :: NonEmpty RcvPublicAuthKey -> Command Recipient
   LSET :: LinkId -> QueueLinkData -> Command Recipient
@@ -566,7 +573,7 @@ data Command (p :: Party) where
   -- SMP notification subscriber commands
   NSUB :: Command Notifier
   -- | subscribe all associated queues. Service ID must be used as entity ID, and service session key must sign the command.
-  NSUBS :: Command NotifierService
+  NSUBS :: Int64 -> IdsHash -> Command NotifierService
   PRXY :: SMPServer -> Maybe BasicAuth -> Command ProxiedClient -- request a relay server connection by URI
   -- Transmission to proxy:
   -- - entity ID: ID of the session with relay returned in PKEY (response to PRXY)
@@ -692,7 +699,7 @@ data BrokerMsg where
   LNK :: SenderId -> QueueLinkData -> BrokerMsg
   -- | Service subscription success - confirms when queue was associated with the service
   SOK :: Maybe ServiceId -> BrokerMsg
-  -- | The number of queues subscribed with SUBS command
+  -- | The number of queues and XOR-hash of their IDs subscribed with SUBS command
   SOKS :: Int64 -> IdsHash -> BrokerMsg
   -- MSG v1/2 has to be supported for encoding/decoding
   -- v1: MSG :: MsgId -> SystemTime -> MsgBody -> BrokerMsg
@@ -1454,7 +1461,44 @@ type MsgId = ByteString
 -- | SMP message body.
 type MsgBody = ByteString
 
-type IdsHash = ByteString
+data ServiceSub = ServiceSub
+  { serviceId :: ServiceId,
+    smpQueueCount :: Int64,
+    smpQueueIdsHash :: IdsHash
+  }
+
+newtype IdsHash = IdsHash {unIdsHash :: BS.ByteString}
+  deriving (Eq, Show)
+  deriving newtype (Encoding, FromField)
+
+instance ToField IdsHash where
+  toField (IdsHash s) = toField (Binary s)
+  {-# INLINE toField #-}
+
+instance Semigroup IdsHash where
+  (IdsHash s1) <> (IdsHash s2) =
+    let !s' = BS.pack $ zipWith xor' (BS.unpack s1) (BS.unpack s2)
+     in IdsHash s'
+
+instance Monoid IdsHash where
+  mempty = IdsHash $ BS.replicate 16 0
+  mconcat ss =
+    let !s' = BS.pack $ foldl' (\ !r (IdsHash s) -> zipWith xor' r (BS.unpack s)) (replicate 16 0) ss -- to prevent packing/unpacking in <> on each step with default mappend
+     in IdsHash s'
+
+xor' :: Word8 -> Word8 -> Word8
+xor' x y = let !r = xor x y in r
+
+noIdsHash ::IdsHash
+noIdsHash = IdsHash B.empty
+{-# INLINE noIdsHash #-}
+
+queueIdsHash :: [QueueId] -> IdsHash
+queueIdsHash = mconcat . map queueIdHash
+
+queueIdHash :: QueueId -> IdsHash
+queueIdHash = IdsHash . C.md5Hash . unEntityId
+{-# INLINE queueIdHash #-}
 
 data ProtocolErrorType = PECmdSyntax | PECmdUnknown | PESession | PEBlock
 
@@ -1668,7 +1712,9 @@ instance PartyI p => ProtocolEncoding SMPVersion ErrorType (Command p) where
         new = e (NEW_, ' ', rKey, dhKey)
         auth = maybe "" (e . ('A',)) auth_
     SUB -> e SUB_
-    SUBS -> e SUBS_
+    SUBS n idsHash
+      | v >= rcvServiceSMPVersion -> e (SUBS_, ' ', n, idsHash)
+      | otherwise -> e SUBS_
     KEY k -> e (KEY_, ' ', k)
     RKEY ks -> e (RKEY_, ' ', ks)
     LSET lnkId d -> e (LSET_, ' ', lnkId, d)
@@ -1684,7 +1730,9 @@ instance PartyI p => ProtocolEncoding SMPVersion ErrorType (Command p) where
     SEND flags msg -> e (SEND_, ' ', flags, ' ', Tail msg)
     PING -> e PING_
     NSUB -> e NSUB_
-    NSUBS -> e NSUBS_
+    NSUBS n idsHash
+      | v >= rcvServiceSMPVersion -> e (NSUBS_, ' ', n, idsHash)
+      | otherwise -> e NSUBS_
     LKEY k -> e (LKEY_, ' ', k)
     LGET -> e LGET_
     PRXY host auth_ -> e (PRXY_, ' ', host, auth_)
@@ -1775,7 +1823,9 @@ instance ProtocolEncoding SMPVersion ErrorType Cmd where
         OFF_ -> pure OFF
         DEL_ -> pure DEL
         QUE_ -> pure QUE
-    CT SRecipientService SUBS_ -> pure $ Cmd SRecipientService SUBS
+    CT SRecipientService SUBS_
+      | v >= rcvServiceSMPVersion -> Cmd SRecipientService <$> (SUBS <$> _smpP <*> smpP)
+      | otherwise -> pure $ Cmd SRecipientService $ SUBS (-1) noIdsHash
     CT SSender tag ->
       Cmd SSender <$> case tag of
         SKEY_ -> SKEY <$> _smpP
@@ -1792,7 +1842,9 @@ instance ProtocolEncoding SMPVersion ErrorType Cmd where
         PFWD_ -> PFWD <$> _smpP <*> smpP <*> (EncTransmission . unTail <$> smpP)
         PRXY_ -> PRXY <$> _smpP <*> smpP
     CT SNotifier NSUB_ -> pure $ Cmd SNotifier NSUB
-    CT SNotifierService NSUBS_ -> pure $ Cmd SNotifierService NSUBS
+    CT SNotifierService NSUBS_
+      | v >= rcvServiceSMPVersion -> Cmd SNotifierService <$> (NSUBS <$> _smpP <*> smpP)
+      | otherwise -> pure $ Cmd SNotifierService $ NSUBS (-1) noIdsHash
 
   fromProtocolError = fromProtocolError @SMPVersion @ErrorType @BrokerMsg
   {-# INLINE fromProtocolError #-}
@@ -1870,7 +1922,7 @@ instance ProtocolEncoding SMPVersion ErrorType BrokerMsg where
     SOK_ -> SOK <$> _smpP
     SOKS_
       | v >= rcvServiceSMPVersion -> SOKS <$> _smpP <*> smpP
-      | otherwise -> SOKS <$> _smpP <*> pure B.empty
+      | otherwise -> SOKS <$> _smpP <*> pure noIdsHash
     NID_ -> NID <$> _smpP <*> smpP
     NMSG_ -> NMSG <$> _smpP <*> smpP
     PKEY_ -> PKEY <$> _smpP <*> smpP <*> smpP

src/Simplex/Messaging/Server.hs

@@ -6,6 +6,7 @@
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE KindSignatures #-}
 {-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE MultiWayIf #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE NumericUnderscores #-}
 {-# LANGUAGE OverloadedLists #-}
@@ -1235,7 +1236,7 @@ verifyQueueTransmission service thAuth (tAuth, authorized, (corrId, entId, comma
     vc SCreator (NEW NewQueueReq {rcvAuthKey = k}) = verifiedWith k
     vc SRecipient SUB = verifyQueue $ \q -> verifiedWithKeys $ recipientKeys (snd q)
     vc SRecipient _ = verifyQueue $ \q -> verifiedWithKeys $ recipientKeys (snd q)
-    vc SRecipientService SUBS = verifyServiceCmd
+    vc SRecipientService SUBS {} = verifyServiceCmd
     vc SSender (SKEY k) = verifySecure k
     -- SEND will be accepted without authorization before the queue is secured with KEY, SKEY or LSKEY command
     vc SSender SEND {} = verifyQueue $ \q -> if maybe (isNothing tAuth) verify (senderKey $ snd q) then VRVerified q_ else VRFailed AUTH
@@ -1243,7 +1244,7 @@ verifyQueueTransmission service thAuth (tAuth, authorized, (corrId, entId, comma
     vc SSenderLink (LKEY k) = verifySecure k
     vc SSenderLink LGET = verifyQueue $ \q -> if isContactQueue (snd q) then VRVerified q_ else VRFailed AUTH
     vc SNotifier NSUB = verifyQueue $ \q -> maybe dummyVerify (\n -> verifiedWith $ notifierKey n) (notifier $ snd q)
-    vc SNotifierService NSUBS = verifyServiceCmd
+    vc SNotifierService NSUBS {} = verifyServiceCmd
     vc SProxiedClient _ = VRVerified Nothing
     vc SProxyService (RFWD _) = VRVerified Nothing
     checkRole = case (service, partyClientRole p) of
@@ -1453,8 +1454,8 @@ client
       Cmd SNotifier NSUB -> response . (corrId,entId,) <$> case q_ of
         Just (q, QueueRec {notifier = Just ntfCreds}) -> subscribeNotifications q ntfCreds
         _ -> pure $ ERR INTERNAL
-      Cmd SNotifierService NSUBS -> response . (corrId,entId,) <$> case clntServiceId of
-        Just serviceId -> subscribeServiceNotifications serviceId
+      Cmd SNotifierService (NSUBS n idsHash) -> response . (corrId,entId,) <$> case clntServiceId of
+        Just serviceId -> subscribeServiceNotifications serviceId (n, idsHash)
         Nothing -> pure $ ERR INTERNAL
       Cmd SCreator (NEW nqr@NewQueueReq {auth_}) ->
         response <$> ifM allowNew (createQueue nqr) (pure (corrId, entId, ERR AUTH))
@@ -1483,8 +1484,8 @@ client
           OFF -> response <$> maybe (pure $ err INTERNAL) suspendQueue_ q_
           DEL -> response <$> maybe (pure $ err INTERNAL) delQueueAndMsgs q_
           QUE -> withQueue $ \q qr -> (corrId,entId,) <$> getQueueInfo q qr
-      Cmd SRecipientService SUBS -> response . (corrId,entId,) <$> case clntServiceId of
-        Just serviceId -> subscribeServiceMessages serviceId
+      Cmd SRecipientService (SUBS n idsHash)-> response . (corrId,entId,) <$> case clntServiceId of
+        Just serviceId -> subscribeServiceMessages serviceId (n, idsHash)
         Nothing -> pure $ ERR INTERNAL -- it's "internal" because it should never get to this branch
       where
         createQueue :: NewQueueReq -> M s (Transmission BrokerMsg)
@@ -1783,9 +1784,9 @@ client
                     TM.insert entId sub $ clientSubs clnt
                     pure (False, Just sub)
 
-        subscribeServiceMessages :: ServiceId -> M s BrokerMsg
-        subscribeServiceMessages serviceId =
-          sharedSubscribeService SRecipientService serviceId subscribers serviceSubscribed serviceSubsCount >>= \case
+        subscribeServiceMessages :: ServiceId -> (Int64, IdsHash) -> M s BrokerMsg
+        subscribeServiceMessages serviceId expected =
+          sharedSubscribeService SRecipientService serviceId expected subscribers serviceSubscribed serviceSubsCount rcvServices >>= \case
             Left e -> pure $ ERR e
             Right (hasSub, (count, idsHash)) -> do
               unless hasSub $ forkClient clnt "deliverServiceMessages" $ liftIO $ deliverServiceMessages count
@@ -1819,25 +1820,33 @@ client
                   TM.insert rId sub $ subscriptions clnt
                   pure $ Just sub
 
-        subscribeServiceNotifications :: ServiceId -> M s BrokerMsg
-        subscribeServiceNotifications serviceId =
-          either ERR (uncurry SOKS . snd) <$> sharedSubscribeService SNotifierService serviceId ntfSubscribers ntfServiceSubscribed ntfServiceSubsCount
+        subscribeServiceNotifications :: ServiceId -> (Int64, IdsHash) -> M s BrokerMsg
+        subscribeServiceNotifications serviceId expected =
+          either ERR (uncurry SOKS . snd) <$> sharedSubscribeService SNotifierService serviceId expected ntfSubscribers ntfServiceSubscribed ntfServiceSubsCount ntfServices
 
-        sharedSubscribeService :: (PartyI p, ServiceParty p) => SParty p -> ServiceId -> ServerSubscribers s -> (Client s -> TVar Bool) -> (Client s -> TVar Int64) -> M s (Either ErrorType (Bool, (Int64, IdsHash)))
-        sharedSubscribeService party serviceId srvSubscribers clientServiceSubscribed clientServiceSubs = do
+        sharedSubscribeService :: (PartyI p, ServiceParty p) => SParty p -> ServiceId -> (Int64, IdsHash) -> ServerSubscribers s -> (Client s -> TVar Bool) -> (Client s -> TVar Int64) -> (ServerStats -> ServiceStats) -> M s (Either ErrorType (Bool, (Int64, IdsHash)))
+        sharedSubscribeService party serviceId (count, idsHash) srvSubscribers clientServiceSubscribed clientServiceSubs servicesSel = do
           subscribed <- readTVarIO $ clientServiceSubscribed clnt
+          stats <- asks serverStats
           liftIO $ runExceptT $
             (subscribed,)
               <$> if subscribed
-                then (,B.empty) <$> readTVarIO (clientServiceSubs clnt) -- TODO [certs rcv] get IDs hash
+                then (,mempty) <$> readTVarIO (clientServiceSubs clnt) -- TODO [certs rcv] get IDs hash
                 else do
-                  count' <- ExceptT $ getServiceQueueCount @(StoreQueue s) (queueStore ms) party serviceId
+                  (count', idsHash') <- ExceptT $ getServiceQueueCountHash @(StoreQueue s) (queueStore ms) party serviceId
                   incCount <- atomically $ do
                     writeTVar (clientServiceSubscribed clnt) True
-                    count <- swapTVar (clientServiceSubs clnt) count'
-                    pure $ count' - count
+                    currCount <- swapTVar (clientServiceSubs clnt) count' -- TODO [certs rcv] maintain IDs hash here?
+                    pure $ count' - currCount
+                  let incSrvStat sel n = liftIO $ atomicModifyIORef'_ (sel $ servicesSel stats) (+ n)
+                      diff = fromIntegral $ count' - count
+                  if -- TODO [certs rcv] account for not provided counts/hashes (expected n = -1)
+                    | diff == 0 && idsHash == idsHash' -> incSrvStat srvSubOk 1
+                    | diff > 0 -> incSrvStat srvSubMore 1 >> incSrvStat srvSubMoreTotal diff
+                    | diff < 0 -> incSrvStat srvSubFewer 1 >> incSrvStat srvSubFewerTotal (- diff)
+                    | otherwise -> incSrvStat srvSubDiff 1
                   atomically $ writeTQueue (subQ srvSubscribers) (CSService serviceId incCount, clientId)
-                  pure (count', B.empty) -- TODO [certs rcv] get IDs hash
+                  pure (count', idsHash')
 
         acknowledgeMsg :: MsgId -> StoreQueue s -> QueueRec -> M s (Transmission BrokerMsg)
         acknowledgeMsg msgId q qr =

src/Simplex/Messaging/Server/MsgStore/Journal.hs

@@ -356,8 +356,8 @@ instance QueueStoreClass (JournalQueue s) (QStore s) where
   {-# INLINE setQueueService #-}
   getQueueNtfServices = withQS (getQueueNtfServices @(JournalQueue s))
   {-# INLINE getQueueNtfServices #-}
-  getServiceQueueCount = withQS (getServiceQueueCount @(JournalQueue s))
-  {-# INLINE getServiceQueueCount #-}
+  getServiceQueueCountHash = withQS (getServiceQueueCountHash @(JournalQueue s))
+  {-# INLINE getServiceQueueCountHash #-}
 
 makeQueue_ :: JournalMsgStore s -> RecipientId -> QueueRec -> Lock -> IO (JournalQueue s)
 makeQueue_ JournalMsgStore {sharedLock} rId qr queueLock = do

src/Simplex/Messaging/Server/Prometheus.hs

@@ -21,6 +21,7 @@ import Simplex.Messaging.Transport (simplexMQVersion)
 import Simplex.Messaging.Transport.Server (SocketStats (..))
 import Simplex.Messaging.Util (tshow)
 
+-- TODO [certs rcv] add service subscriptions and count/hash diffs
 data ServerMetrics = ServerMetrics
   { statsData :: ServerStatsData,
     activeQueueCounts :: PeriodStatCounts,

src/Simplex/Messaging/Server/QueueStore/Postgres.hs

@@ -485,15 +485,11 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where
         let (sNtfs, restNtfs) = partition (\(nId, _) -> S.member nId snIds) ntfs'
          in ((serviceId, sNtfs) : ssNtfs, restNtfs)
 
-  getServiceQueueCount :: (PartyI p, ServiceParty p) => PostgresQueueStore q -> SParty p -> ServiceId -> IO (Either ErrorType Int64)
-  getServiceQueueCount st party serviceId =
-    E.uninterruptibleMask_ $ runExceptT $ withDB' "getServiceQueueCount" st $ \db ->
-      fmap (fromMaybe 0) $ maybeFirstRow fromOnly $
-        DB.query db query (Only serviceId)
-    where
-      query = case party of
-        SRecipientService -> "SELECT count(1) FROM msg_queues WHERE rcv_service_id = ? AND deleted_at IS NULL"
-        SNotifierService -> "SELECT count(1) FROM msg_queues WHERE ntf_service_id = ? AND deleted_at IS NULL"
+  getServiceQueueCountHash :: (PartyI p, ServiceParty p) => PostgresQueueStore q -> SParty p -> ServiceId -> IO (Either ErrorType (Int64, IdsHash))
+  getServiceQueueCountHash st party serviceId =
+    E.uninterruptibleMask_ $ runExceptT $ withDB' "getServiceQueueCountHash" st $ \db ->
+      fmap (fromMaybe (0, mempty)) $ maybeFirstRow id $
+        DB.query db ("SELECT queue_count, queue_ids_hash FROM services WHERE service_id = ? AND service_role = ?") (serviceId, partyServiceRole party)
 
 batchInsertServices :: [STMService] -> PostgresQueueStore q -> IO Int64
 batchInsertServices services' toStore =
@@ -740,6 +736,10 @@ instance ToField C.APublicAuthKey where toField = toField . Binary . C.encodePub
 
 instance FromField C.APublicAuthKey where fromField = blobFieldDecoder C.decodePubKey
 
+instance ToField IdsHash where toField (IdsHash s) = toField (Binary s)
+
+deriving newtype instance FromField IdsHash
+
 instance ToField EncDataBytes where toField (EncDataBytes s) = toField (Binary s)
 
 deriving newtype instance FromField EncDataBytes

src/Simplex/Messaging/Server/QueueStore/Postgres/Migrations.hs

@@ -166,27 +166,33 @@ m20250830_queue_ids_hash =
   createXorHashFuncs
     <> T.pack
       [r|
-ALTER TABLE services ADD COLUMN queue_ids_hash BYTEA NOT NULL DEFAULT '\x00000000000000000000000000000000';
+ALTER TABLE services
+  ADD COLUMN queue_count BIGINT NOT NULL DEFAULT 0,
+  ADD COLUMN queue_ids_hash BYTEA NOT NULL DEFAULT '\x00000000000000000000000000000000';
 
-WITH hashes AS (
+WITH acc AS (
   SELECT
     s.service_id,
-    xor_aggregate(public.digest(CASE WHEN s.service_role = 'M' THEN q.recipient_id ELSE COALESCE(q.notifier_id, '\x00000000000000000000000000000000') END, 'md5')) AS computed_hash
+    count(1) as q_count,
+    xor_aggregate(public.digest(CASE WHEN s.service_role = 'M' THEN q.recipient_id ELSE COALESCE(q.notifier_id, '\x00000000000000000000000000000000') END, 'md5')) AS q_ids_hash
   FROM services s
   JOIN msg_queues q ON (s.service_id = q.rcv_service_id AND s.service_role = 'M') OR (s.service_id = q.ntf_service_id AND s.service_role = 'N')
+  WHERE q.deleted_at IS NULL
   GROUP BY s.service_id
 )
 UPDATE services s
-SET queue_ids_hash = COALESCE(h.computed_hash, '\x00000000000000000000000000000000')
-FROM hashes h
-WHERE s.service_id = h.service_id;
+SET queue_count = COALESCE(acc.q_count, 0),
+    queue_ids_hash = COALESCE(acc.q_ids_hash, '\x00000000000000000000000000000000')
+FROM acc
+WHERE s.service_id = acc.service_id;
 
-CREATE OR REPLACE FUNCTION update_ids_hash(p_service_id BYTEA, p_role text, p_queue_id BYTEA) RETURNS void
+CREATE OR REPLACE FUNCTION update_ids_hash(p_service_id BYTEA, p_role TEXT, p_queue_id BYTEA, p_change BIGINT) RETURNS void
 LANGUAGE plpgsql
 AS $$
 BEGIN
   UPDATE services
-  SET queue_ids_hash = xor_combine(queue_ids_hash, public.digest(p_queue_id, 'md5'))
+  SET queue_count = queue_count + p_change,
+      queue_ids_hash = xor_combine(queue_ids_hash, public.digest(p_queue_id, 'md5'))
   WHERE service_id = p_service_id AND service_role = p_role;
 END;
 $$;
@@ -196,10 +202,10 @@ LANGUAGE plpgsql
 AS $$
 BEGIN
   IF NEW.rcv_service_id IS NOT NULL THEN
-    PERFORM update_ids_hash(NEW.rcv_service_id, 'M', NEW.recipient_id);
+    PERFORM update_ids_hash(NEW.rcv_service_id, 'M', NEW.recipient_id, 1);
   END IF;
-  IF NEW.ntf_service_id IS NOT NULL THEN
-    PERFORM update_ids_hash(NEW.ntf_service_id, 'N', NEW.notifier_id);
+  IF NEW.ntf_service_id IS NOT NULL AND NEW.notifier_id IS NOT NULL THEN
+    PERFORM update_ids_hash(NEW.ntf_service_id, 'N', NEW.notifier_id, 1);
   END IF;
   RETURN NEW;
 END;
@@ -209,11 +215,13 @@ CREATE OR REPLACE FUNCTION on_queue_delete() RETURNS TRIGGER
 LANGUAGE plpgsql
 AS $$
 BEGIN
-  IF OLD.rcv_service_id IS NOT NULL THEN
-    PERFORM update_ids_hash(OLD.rcv_service_id, 'M', OLD.recipient_id);
-  END IF;
-  IF OLD.ntf_service_id IS NOT NULL THEN
-    PERFORM update_ids_hash(OLD.ntf_service_id, 'N', OLD.notifier_id);
+  IF OLD.deleted_at IS NULL THEN
+    IF OLD.rcv_service_id IS NOT NULL THEN
+      PERFORM update_ids_hash(OLD.rcv_service_id, 'M', OLD.recipient_id, -1);
+    END IF;
+    IF OLD.ntf_service_id IS NOT NULL AND OLD.notifier_id IS NOT NULL THEN
+      PERFORM update_ids_hash(OLD.ntf_service_id, 'N', OLD.notifier_id, -1);
+    END IF;
   END IF;
   RETURN OLD;
 END;
@@ -223,21 +231,26 @@ CREATE OR REPLACE FUNCTION on_queue_update() RETURNS TRIGGER
 LANGUAGE plpgsql
 AS $$
 BEGIN
-  IF OLD.rcv_service_id IS DISTINCT FROM NEW.rcv_service_id THEN
-    IF OLD.rcv_service_id IS NOT NULL THEN
-      PERFORM update_ids_hash(OLD.rcv_service_id, 'M', OLD.recipient_id);
-    END IF;
-    IF NEW.rcv_service_id IS NOT NULL THEN
-      PERFORM update_ids_hash(NEW.rcv_service_id, 'M', NEW.recipient_id);
+  IF OLD.deleted_at IS NULL AND OLD.rcv_service_id IS NOT NULL THEN
+    IF NOT (NEW.deleted_at IS NULL AND NEW.rcv_service_id IS NOT NULL) THEN
+      PERFORM update_ids_hash(OLD.rcv_service_id, 'M', OLD.recipient_id, -1);
+    ELSIF OLD.rcv_service_id IS DISTINCT FROM NEW.rcv_service_id THEN
+      PERFORM update_ids_hash(OLD.rcv_service_id, 'M', OLD.recipient_id, -1);
+      PERFORM update_ids_hash(NEW.rcv_service_id, 'M', NEW.recipient_id, 1);
     END IF;
+  ELSIF NEW.deleted_at IS NULL AND NEW.rcv_service_id IS NOT NULL THEN
+    PERFORM update_ids_hash(NEW.rcv_service_id, 'M', NEW.recipient_id, 1);
   END IF;
-  IF OLD.ntf_service_id IS DISTINCT FROM NEW.ntf_service_id THEN
-    IF OLD.ntf_service_id IS NOT NULL THEN
-      PERFORM update_ids_hash(OLD.ntf_service_id, 'N', OLD.notifier_id);
-    END IF;
-    IF NEW.ntf_service_id IS NOT NULL THEN
-      PERFORM update_ids_hash(NEW.ntf_service_id, 'N', NEW.notifier_id);
+
+  IF OLD.deleted_at IS NULL AND OLD.ntf_service_id IS NOT NULL AND OLD.notifier_id IS NOT NULL THEN
+    IF NOT (NEW.deleted_at IS NULL AND NEW.ntf_service_id IS NOT NULL AND NEW.notifier_id IS NOT NULL) THEN
+      PERFORM update_ids_hash(OLD.ntf_service_id, 'N', OLD.notifier_id, -1);
+    ELSIF OLD.ntf_service_id IS DISTINCT FROM NEW.ntf_service_id OR OLD.notifier_id IS DISTINCT FROM NEW.notifier_id THEN
+      PERFORM update_ids_hash(OLD.ntf_service_id, 'N', OLD.notifier_id, -1);
+      PERFORM update_ids_hash(NEW.ntf_service_id, 'N', NEW.notifier_id, 1);
     END IF;
+  ELSIF NEW.deleted_at IS NULL AND NEW.ntf_service_id IS NOT NULL AND NEW.notifier_id IS NOT NULL THEN
+    PERFORM update_ids_hash(NEW.ntf_service_id, 'N', NEW.notifier_id, 1);
   END IF;
   RETURN NEW;
 END;
@@ -274,7 +287,9 @@ DROP FUNCTION on_queue_update;
 
 DROP FUNCTION update_ids_hash;
 
-ALTER TABLE services DROP COLUMN queue_ids_hash;
+ALTER TABLE services
+  DROP COLUMN queue_count,
+  DROP COLUMN queue_ids_hash;
     |]
     <> dropXorHashFuncs
 

src/Simplex/Messaging/Server/QueueStore/Postgres/server_schema.sql

@@ -20,11 +20,13 @@ CREATE FUNCTION smp_server.on_queue_delete() RETURNS trigger
     LANGUAGE plpgsql
     AS $$
 BEGIN
-  IF OLD.rcv_service_id IS NOT NULL THEN
-    PERFORM update_ids_hash(OLD.rcv_service_id, 'M', OLD.recipient_id);
-  END IF;
-  IF OLD.ntf_service_id IS NOT NULL THEN
-    PERFORM update_ids_hash(OLD.ntf_service_id, 'N', OLD.notifier_id);
+  IF OLD.deleted_at IS NULL THEN
+    IF OLD.rcv_service_id IS NOT NULL THEN
+      PERFORM update_ids_hash(OLD.rcv_service_id, 'M', OLD.recipient_id, -1);
+    END IF;
+    IF OLD.ntf_service_id IS NOT NULL AND OLD.notifier_id IS NOT NULL THEN
+      PERFORM update_ids_hash(OLD.ntf_service_id, 'N', OLD.notifier_id, -1);
+    END IF;
   END IF;
   RETURN OLD;
 END;
@@ -37,10 +39,10 @@ CREATE FUNCTION smp_server.on_queue_insert() RETURNS trigger
     AS $$
 BEGIN
   IF NEW.rcv_service_id IS NOT NULL THEN
-    PERFORM update_ids_hash(NEW.rcv_service_id, 'M', NEW.recipient_id);
+    PERFORM update_ids_hash(NEW.rcv_service_id, 'M', NEW.recipient_id, 1);
   END IF;
-  IF NEW.ntf_service_id IS NOT NULL THEN
-    PERFORM update_ids_hash(NEW.ntf_service_id, 'N', NEW.notifier_id);
+  IF NEW.ntf_service_id IS NOT NULL AND NEW.notifier_id IS NOT NULL THEN
+    PERFORM update_ids_hash(NEW.ntf_service_id, 'N', NEW.notifier_id, 1);
   END IF;
   RETURN NEW;
 END;
@@ -52,21 +54,26 @@ CREATE FUNCTION smp_server.on_queue_update() RETURNS trigger
     LANGUAGE plpgsql
     AS $$
 BEGIN
-  IF OLD.rcv_service_id IS DISTINCT FROM NEW.rcv_service_id THEN
-    IF OLD.rcv_service_id IS NOT NULL THEN
-      PERFORM update_ids_hash(OLD.rcv_service_id, 'M', OLD.recipient_id);
-    END IF;
-    IF NEW.rcv_service_id IS NOT NULL THEN
-      PERFORM update_ids_hash(NEW.rcv_service_id, 'M', NEW.recipient_id);
+  IF OLD.deleted_at IS NULL AND OLD.rcv_service_id IS NOT NULL THEN
+    IF NOT (NEW.deleted_at IS NULL AND NEW.rcv_service_id IS NOT NULL) THEN
+      PERFORM update_ids_hash(OLD.rcv_service_id, 'M', OLD.recipient_id, -1);
+    ELSIF OLD.rcv_service_id IS DISTINCT FROM NEW.rcv_service_id THEN
+      PERFORM update_ids_hash(OLD.rcv_service_id, 'M', OLD.recipient_id, -1);
+      PERFORM update_ids_hash(NEW.rcv_service_id, 'M', NEW.recipient_id, 1);
     END IF;
+  ELSIF NEW.deleted_at IS NULL AND NEW.rcv_service_id IS NOT NULL THEN
+    PERFORM update_ids_hash(NEW.rcv_service_id, 'M', NEW.recipient_id, 1);
   END IF;
-  IF OLD.ntf_service_id IS DISTINCT FROM NEW.ntf_service_id THEN
-    IF OLD.ntf_service_id IS NOT NULL THEN
-      PERFORM update_ids_hash(OLD.ntf_service_id, 'N', OLD.notifier_id);
-    END IF;
-    IF NEW.ntf_service_id IS NOT NULL THEN
-      PERFORM update_ids_hash(NEW.ntf_service_id, 'N', NEW.notifier_id);
+
+  IF OLD.deleted_at IS NULL AND OLD.ntf_service_id IS NOT NULL AND OLD.notifier_id IS NOT NULL THEN
+    IF NOT (NEW.deleted_at IS NULL AND NEW.ntf_service_id IS NOT NULL AND NEW.notifier_id IS NOT NULL) THEN
+      PERFORM update_ids_hash(OLD.ntf_service_id, 'N', OLD.notifier_id, -1);
+    ELSIF OLD.ntf_service_id IS DISTINCT FROM NEW.ntf_service_id OR OLD.notifier_id IS DISTINCT FROM NEW.notifier_id THEN
+      PERFORM update_ids_hash(OLD.ntf_service_id, 'N', OLD.notifier_id, -1);
+      PERFORM update_ids_hash(NEW.ntf_service_id, 'N', NEW.notifier_id, 1);
     END IF;
+  ELSIF NEW.deleted_at IS NULL AND NEW.ntf_service_id IS NOT NULL AND NEW.notifier_id IS NOT NULL THEN
+    PERFORM update_ids_hash(NEW.ntf_service_id, 'N', NEW.notifier_id, 1);
   END IF;
   RETURN NEW;
 END;
@@ -74,12 +81,13 @@ $$;
 
 
 
-CREATE FUNCTION smp_server.update_ids_hash(p_service_id bytea, p_role text, p_queue_id bytea) RETURNS void
+CREATE FUNCTION smp_server.update_ids_hash(p_service_id bytea, p_role text, p_queue_id bytea, p_change bigint) RETURNS void
     LANGUAGE plpgsql
     AS $$
 BEGIN
   UPDATE services
-  SET queue_ids_hash = xor_combine(queue_ids_hash, public.digest(p_queue_id, 'md5'))
+  SET queue_count = queue_count + p_change,
+      queue_ids_hash = xor_combine(queue_ids_hash, public.digest(p_queue_id, 'md5'))
   WHERE service_id = p_service_id AND service_role = p_role;
 END;
 $$;
@@ -152,6 +160,7 @@ CREATE TABLE smp_server.services (
     service_cert bytea NOT NULL,
     service_cert_hash bytea NOT NULL,
     created_at bigint NOT NULL,
+    queue_count bigint DEFAULT 0 NOT NULL,
     queue_ids_hash bytea DEFAULT '\x00000000000000000000000000000000'::bytea NOT NULL
 );
 

src/Simplex/Messaging/Server/QueueStore/STM.hs

@@ -28,6 +28,7 @@ where
 import qualified Control.Exception as E
 import Control.Logger.Simple
 import Control.Monad
+import Data.Bifunctor (first)
 import Data.Bitraversable (bimapM)
 import Data.Functor (($>))
 import Data.Int (Int64)
@@ -61,8 +62,8 @@ data STMQueueStore q = STMQueueStore
 
 data STMService = STMService
   { serviceRec :: ServiceRec,
-    serviceRcvQueues :: TVar (Set RecipientId),
-    serviceNtfQueues :: TVar (Set NotifierId)
+    serviceRcvQueues :: TVar (Set RecipientId, IdsHash), -- TODO [certs rcv] get/maintain hash
+    serviceNtfQueues :: TVar (Set NotifierId, IdsHash) -- TODO [certs rcv] get/maintain hash
   }
 
 setStoreLog :: STMQueueStore q -> StoreLog 'WriteMode -> IO ()
@@ -112,7 +113,7 @@ instance StoreQueueClass q => QueueStoreClass q (STMQueueStore q) where
         }
     where
       serviceCount role = M.foldl' (\ !n s -> if serviceRole (serviceRec s) == role then n + 1 else n) 0
-      serviceQueuesCount serviceSel = foldM (\n s -> (n +) . S.size <$> readTVarIO (serviceSel s)) 0
+      serviceQueuesCount serviceSel = foldM (\n s -> (n +) . S.size . fst <$> readTVarIO (serviceSel s)) 0
 
   addQueue_ :: STMQueueStore q -> (RecipientId -> QueueRec -> IO q) -> RecipientId -> QueueRec -> IO (Either ErrorType q)
   addQueue_ st mkQ rId qr@QueueRec {senderId = sId, notifier, queueData, rcvServiceId} = do
@@ -302,8 +303,8 @@ instance StoreQueueClass q => QueueStoreClass q (STMQueueStore q) where
         TM.insert fp newSrvId serviceCerts
         pure $ Right (newSrvId, True)
       newSTMService = do
-        serviceRcvQueues <- newTVar S.empty
-        serviceNtfQueues <- newTVar S.empty
+        serviceRcvQueues <- newTVar (S.empty, mempty)
+        serviceNtfQueues <- newTVar (S.empty, mempty)
         pure STMService {serviceRec = sr, serviceRcvQueues, serviceNtfQueues}
 
   setQueueService :: (PartyI p, ServiceParty p) => STMQueueStore q -> q -> SParty p -> Maybe ServiceId -> IO (Either ErrorType ())
@@ -329,7 +330,7 @@ instance StoreQueueClass q => QueueStoreClass q (STMQueueStore q) where
                 let !q' = Just q {notifier = Just nc {ntfServiceId = serviceId}}
                 updateServiceQueues serviceNtfQueues nId prevNtfSrvId
                 writeTVar qr q' $> Right ()
-      updateServiceQueues :: (STMService -> TVar (Set QueueId)) -> QueueId -> Maybe ServiceId -> STM ()
+      updateServiceQueues :: (STMService -> TVar (Set QueueId, IdsHash)) -> QueueId -> Maybe ServiceId -> STM ()
       updateServiceQueues serviceSel qId prevSrvId = do
         mapM_ (removeServiceQueue st serviceSel qId) prevSrvId
         mapM_ (addServiceQueue st serviceSel qId) serviceId
@@ -344,16 +345,16 @@ instance StoreQueueClass q => QueueStoreClass q (STMQueueStore q) where
     pure $ Right (ssNtfs', deleteNtfs)
     where
       addService (ssNtfs, ntfs') (serviceId, s) = do
-        snIds <- readTVarIO $ serviceNtfQueues s
+        (snIds, _) <- readTVarIO $ serviceNtfQueues s
         let (sNtfs, restNtfs) = partition (\(nId, _) -> S.member nId snIds) ntfs'
         pure ((Just serviceId, sNtfs) : ssNtfs, restNtfs)
 
-  getServiceQueueCount :: (PartyI p, ServiceParty p) => STMQueueStore q -> SParty p -> ServiceId -> IO (Either ErrorType Int64)
-  getServiceQueueCount st party serviceId =
+  getServiceQueueCountHash :: (PartyI p, ServiceParty p) => STMQueueStore q -> SParty p -> ServiceId -> IO (Either ErrorType (Int64, IdsHash))
+  getServiceQueueCountHash st party serviceId =
     TM.lookupIO serviceId (services st) >>=
-      maybe (pure $ Left AUTH) (fmap (Right . fromIntegral . S.size) . readTVarIO . serviceSel)
+      maybe (pure $ Left AUTH) (fmap (Right . first (fromIntegral . S.size)) . readTVarIO . serviceSel)
     where
-      serviceSel :: STMService -> TVar (Set QueueId)
+      serviceSel :: STMService -> TVar (Set QueueId, IdsHash)
       serviceSel = case party of
         SRecipientService -> serviceRcvQueues
         SNotifierService -> serviceNtfQueues
@@ -364,7 +365,7 @@ foldRcvServiceQueues st serviceId f acc =
     Nothing -> pure acc
     Just s ->
       readTVarIO (serviceRcvQueues s)
-        >>= foldM (\a -> get >=> maybe (pure a) (f a)) acc
+        >>= foldM (\a -> get >=> maybe (pure a) (f a)) acc . fst
   where
     get rId = TM.lookupIO rId (queues st) $>>= \q -> (q,) <$$> readTVarIO (queueRec q)
 
@@ -377,16 +378,23 @@ setStatus qr status =
     Just q -> (Right (), Just q {status})
     Nothing -> (Left AUTH, Nothing)
 
-addServiceQueue :: STMQueueStore q -> (STMService -> TVar (Set QueueId)) -> QueueId -> ServiceId -> STM ()
-addServiceQueue st serviceSel qId serviceId =
-  TM.lookup serviceId (services st) >>= mapM_ (\s -> modifyTVar' (serviceSel s) (S.insert qId))
+addServiceQueue :: STMQueueStore q -> (STMService -> TVar (Set QueueId, IdsHash)) -> QueueId -> ServiceId -> STM ()
+addServiceQueue = setServiceQueues_ S.insert
 {-# INLINE addServiceQueue #-}
 
-removeServiceQueue :: STMQueueStore q -> (STMService -> TVar (Set QueueId)) -> QueueId -> ServiceId -> STM ()
-removeServiceQueue st serviceSel qId serviceId =
-  TM.lookup serviceId (services st) >>= mapM_ (\s -> modifyTVar' (serviceSel s) (S.delete qId))
+removeServiceQueue :: STMQueueStore q -> (STMService -> TVar (Set QueueId, IdsHash)) -> QueueId -> ServiceId -> STM ()
+removeServiceQueue = setServiceQueues_ S.delete
 {-# INLINE removeServiceQueue #-}
 
+setServiceQueues_ :: (QueueId -> Set QueueId -> Set QueueId) -> STMQueueStore q -> (STMService -> TVar (Set QueueId, IdsHash)) -> QueueId -> ServiceId -> STM ()
+setServiceQueues_ updateSet st serviceSel qId serviceId =
+  TM.lookup serviceId (services st) >>= mapM_ (\v -> modifyTVar' (serviceSel v) update)
+  where
+    update (s, idsHash) =
+      let !s' = updateSet qId s
+          !idsHash' = queueIdHash qId <> idsHash
+       in (s', idsHash')
+
 removeNotifier :: STMQueueStore q -> NtfCreds -> STM ()
 removeNotifier st NtfCreds {notifierId = nId, ntfServiceId} = do
   TM.delete nId $ notifiers st

src/Simplex/Messaging/Server/QueueStore/Types.hs

@@ -48,7 +48,7 @@ class StoreQueueClass q => QueueStoreClass q s where
   getCreateService :: s -> ServiceRec -> IO (Either ErrorType ServiceId)
   setQueueService :: (PartyI p, ServiceParty p) => s -> q -> SParty p -> Maybe ServiceId -> IO (Either ErrorType ())
   getQueueNtfServices :: s -> [(NotifierId, a)] -> IO (Either ErrorType ([(Maybe ServiceId, [(NotifierId, a)])], [(NotifierId, a)]))
-  getServiceQueueCount :: (PartyI p, ServiceParty p) => s -> SParty p -> ServiceId -> IO (Either ErrorType Int64)
+  getServiceQueueCountHash :: (PartyI p, ServiceParty p) => s -> SParty p -> ServiceId -> IO (Either ErrorType (Int64, IdsHash))
 
 data EntityCounts = EntityCounts
   { queueCount :: Int,

src/Simplex/Messaging/Server/Stats.hs

@@ -821,7 +821,15 @@ data ServiceStats = ServiceStats
     srvSubCount :: IORef Int,
     srvSubDuplicate :: IORef Int,
     srvSubQueues :: IORef Int,
-    srvSubEnd :: IORef Int
+    srvSubEnd :: IORef Int,
+    -- counts of subscriptions
+    srvSubOk :: IORef Int, -- server has the same queues as expected
+    srvSubMore :: IORef Int, -- server has more queues than expected
+    srvSubFewer :: IORef Int, -- server has fewer queues than expected
+    srvSubDiff :: IORef Int, -- server has the same count, but different queues than expected (based on xor-hash)
+    -- adds actual deviations
+    srvSubMoreTotal :: IORef Int, -- server has more queues than expected, adds diff
+    srvSubFewerTotal :: IORef Int
   }
 
 data ServiceStatsData = ServiceStatsData
@@ -832,7 +840,13 @@ data ServiceStatsData = ServiceStatsData
     _srvSubCount :: Int,
     _srvSubDuplicate :: Int,
     _srvSubQueues :: Int,
-    _srvSubEnd :: Int
+    _srvSubEnd :: Int,
+    _srvSubOk :: Int,
+    _srvSubMore :: Int,
+    _srvSubFewer :: Int,
+    _srvSubDiff :: Int,
+    _srvSubMoreTotal :: Int,
+    _srvSubFewerTotal :: Int
   }
   deriving (Show)
 
@@ -846,7 +860,13 @@ newServiceStatsData =
       _srvSubCount = 0,
       _srvSubDuplicate = 0,
       _srvSubQueues = 0,
-      _srvSubEnd = 0
+      _srvSubEnd = 0,
+      _srvSubOk = 0,
+      _srvSubMore = 0,
+      _srvSubFewer = 0,
+      _srvSubDiff = 0,
+      _srvSubMoreTotal = 0,
+      _srvSubFewerTotal = 0
     }
 
 newServiceStats :: IO ServiceStats
@@ -859,6 +879,12 @@ newServiceStats = do
   srvSubDuplicate <- newIORef 0
   srvSubQueues <- newIORef 0
   srvSubEnd <- newIORef 0
+  srvSubOk <- newIORef 0
+  srvSubMore <- newIORef 0
+  srvSubFewer <- newIORef 0
+  srvSubDiff <- newIORef 0
+  srvSubMoreTotal <- newIORef 0
+  srvSubFewerTotal <- newIORef 0
   pure
     ServiceStats
       { srvAssocNew,
@@ -868,7 +894,13 @@ newServiceStats = do
         srvSubCount,
         srvSubDuplicate,
         srvSubQueues,
-        srvSubEnd
+        srvSubEnd,
+        srvSubOk,
+        srvSubMore,
+        srvSubFewer,
+        srvSubDiff,
+        srvSubMoreTotal,
+        srvSubFewerTotal
       }
 
 getServiceStatsData :: ServiceStats -> IO ServiceStatsData
@@ -881,6 +913,12 @@ getServiceStatsData s = do
   _srvSubDuplicate <- readIORef $ srvSubDuplicate s
   _srvSubQueues <- readIORef $ srvSubQueues s
   _srvSubEnd <- readIORef $ srvSubEnd s
+  _srvSubOk <- readIORef $ srvSubOk s
+  _srvSubMore <- readIORef $ srvSubMore s
+  _srvSubFewer <- readIORef $ srvSubFewer s
+  _srvSubDiff <- readIORef $ srvSubDiff s
+  _srvSubMoreTotal <- readIORef $ srvSubMoreTotal s
+  _srvSubFewerTotal <- readIORef $ srvSubFewerTotal s
   pure
     ServiceStatsData
       { _srvAssocNew,
@@ -890,7 +928,13 @@ getServiceStatsData s = do
         _srvSubCount,
         _srvSubDuplicate,
         _srvSubQueues,
-        _srvSubEnd
+        _srvSubEnd,
+        _srvSubOk,
+        _srvSubMore,
+        _srvSubFewer,
+        _srvSubDiff,
+        _srvSubMoreTotal,
+        _srvSubFewerTotal
       }
 
 getResetServiceStatsData :: ServiceStats -> IO ServiceStatsData
@@ -903,6 +947,12 @@ getResetServiceStatsData s = do
   _srvSubDuplicate <- atomicSwapIORef (srvSubDuplicate s) 0
   _srvSubQueues <- atomicSwapIORef (srvSubQueues s) 0
   _srvSubEnd <- atomicSwapIORef (srvSubEnd s) 0
+  _srvSubOk <- atomicSwapIORef (srvSubOk s) 0
+  _srvSubMore <- atomicSwapIORef (srvSubMore s) 0
+  _srvSubFewer <- atomicSwapIORef (srvSubFewer s) 0
+  _srvSubDiff <- atomicSwapIORef (srvSubDiff s) 0
+  _srvSubMoreTotal <- atomicSwapIORef (srvSubMoreTotal s) 0
+  _srvSubFewerTotal <- atomicSwapIORef (srvSubFewerTotal s) 0
   pure
     ServiceStatsData
       { _srvAssocNew,
@@ -912,7 +962,13 @@ getResetServiceStatsData s = do
         _srvSubCount,
         _srvSubDuplicate,
         _srvSubQueues,
-        _srvSubEnd
+        _srvSubEnd,
+        _srvSubOk,
+        _srvSubMore,
+        _srvSubFewer,
+        _srvSubDiff,
+        _srvSubMoreTotal,
+        _srvSubFewerTotal
       }
 
 -- this function is not thread safe, it is used on server start only
@@ -926,6 +982,12 @@ setServiceStats s d = do
   writeIORef (srvSubDuplicate s) $! _srvSubDuplicate d
   writeIORef (srvSubQueues s) $! _srvSubQueues d
   writeIORef (srvSubEnd s) $! _srvSubEnd d
+  writeIORef (srvSubOk s) $! _srvSubOk d
+  writeIORef (srvSubMore s) $! _srvSubMore d
+  writeIORef (srvSubFewer s) $! _srvSubFewer d
+  writeIORef (srvSubDiff s) $! _srvSubDiff d
+  writeIORef (srvSubMoreTotal s) $! _srvSubMoreTotal d
+  writeIORef (srvSubFewerTotal s) $! _srvSubFewerTotal d
 
 instance StrEncoding ServiceStatsData where
   strEncode ServiceStatsData {_srvAssocNew, _srvAssocDuplicate, _srvAssocUpdated, _srvAssocRemoved, _srvSubCount, _srvSubDuplicate, _srvSubQueues, _srvSubEnd} =
@@ -963,7 +1025,13 @@ instance StrEncoding ServiceStatsData where
           _srvSubCount,
           _srvSubDuplicate,
           _srvSubQueues,
-          _srvSubEnd
+          _srvSubEnd,
+          _srvSubOk = 0,
+          _srvSubMore = 0,
+          _srvSubFewer = 0,
+          _srvSubDiff = 0,
+          _srvSubMoreTotal = 0,
+          _srvSubFewerTotal = 0
         }
 
 data TimeBuckets = TimeBuckets

src/Simplex/Messaging/Server/StoreLog/ReadWrite.hs

@@ -61,7 +61,7 @@ readQueueStore tty mkQ f st = readLogLines tty f $ \_ -> processLine
             Left e -> logError $ errPfx <> tshow e
             where
               errPfx = "STORE: getCreateService, stored service " <> decodeLatin1 (strEncode serviceId) <> ", "
-          QueueService rId (ASP party) serviceId -> withQueue rId "QueueService" $ \q -> setQueueService st q party serviceId
+          QueueService qId (ASP party) serviceId -> withQueue qId "QueueService" $ \q -> setQueueService st q party serviceId
         printError :: String -> IO ()
         printError e = B.putStrLn $ "Error parsing log: " <> B.pack e <> " - " <> s
         withQueue :: forall a. RecipientId -> T.Text -> (q -> IO (Either ErrorType a)) -> IO ()

tests/ServerTests.hs

@@ -705,15 +705,17 @@ testServiceDeliverSubscribe =
         pure (rId, sId, dec, serviceId)
 
       runSMPServiceClient t (tlsCred, serviceKeys) $ \sh -> do
-        Resp "10" NoEntity (ERR (CMD NO_AUTH)) <- signSendRecv sh aServicePK ("10", NoEntity, SUBS)
-        signSend_ sh aServicePK Nothing ("11", serviceId, SUBS)
+        let idsHash = queueIdsHash [rId]
+        Resp "10" NoEntity (ERR (CMD NO_AUTH)) <- signSendRecv sh aServicePK ("10", NoEntity, SUBS 1 idsHash)
+        signSend_ sh aServicePK Nothing ("11", serviceId, SUBS 1 idsHash)  -- TODO [certs rcv] compute and compare hashes
         [mId3] <-
           fmap catMaybes $
             receiveInAnyOrder -- race between SOKS and MSG, clients can handle it
               sh
               [ \case
-                  Resp "11" serviceId' (SOKS n _) -> do
+                  Resp "11" serviceId' (SOKS n idsHash') -> do
                     n `shouldBe` 1
+                    idsHash' `shouldBe` idsHash
                     serviceId' `shouldBe` serviceId
                     pure $ Just Nothing
                   _ -> pure Nothing,
@@ -798,14 +800,16 @@ testServiceUpgradeAndDowngrade =
       Resp "12" _ OK <- signSendRecv h sKey2 ("12", sId2, _SEND "hello 3.2")
 
       runSMPServiceClient t (tlsCred, serviceKeys) $ \sh -> do
-        signSend_ sh aServicePK Nothing ("14", serviceId, SUBS)
+        let idsHash = queueIdsHash [rId, rId2, rId3]
+        signSend_ sh aServicePK Nothing ("14", serviceId, SUBS 3 idsHash) -- TODO [certs rcv] compute hash
         [(rKey3_1, rId3_1, mId3_1), (rKey3_2, rId3_2, mId3_2)] <-
           fmap catMaybes $
             receiveInAnyOrder -- race between SOKS and MSG, clients can handle it
               sh
               [ \case
-                  Resp "14" serviceId' (SOKS n _) -> do
+                  Resp "14" serviceId' (SOKS n idsHash') -> do
                     n `shouldBe` 3
+                    idsHash' `shouldBe` idsHash
                     serviceId' `shouldBe` serviceId
                     pure $ Just Nothing
                   _ -> pure Nothing,
@@ -828,7 +832,7 @@ testServiceUpgradeAndDowngrade =
       Resp "17" _ OK <- signSendRecv h sKey ("17", sId, _SEND "hello 4")
 
       runSMPClient t $ \sh -> do
-        Resp "18" _ (ERR SERVICE) <- signSendRecv sh aServicePK ("18", serviceId, SUBS)
+        Resp "18" _ (ERR SERVICE) <- signSendRecv sh aServicePK ("18", serviceId, SUBS 3 mempty)
         (Resp "19" rId' (SOK Nothing), Resp "" rId'' (Msg mId4 msg4)) <- signSendRecv2 sh rKey ("19", rId, SUB)
         rId' `shouldBe` rId
         rId'' `shouldBe` rId
@@ -1347,7 +1351,9 @@ testMessageServiceNotifications =
           deliverMessage rh rId rKey sh sId sKey nh2 "connection 1" dec
           deliverMessage rh rId'' rKey'' sh sId'' sKey'' nh2 "connection 2" dec''
           -- -- another client makes service subscription
-          Resp "12" serviceId5 (SOKS 2 _) <- signSendRecv nh1 (C.APrivateAuthKey C.SEd25519 servicePK) ("12", serviceId, NSUBS)
+          let idsHash = queueIdsHash [nId', nId'']
+          Resp "12" serviceId5 (SOKS 2 idsHash') <- signSendRecv nh1 (C.APrivateAuthKey C.SEd25519 servicePK) ("12", serviceId, NSUBS 2 idsHash) -- TODO [certs rcv] compute and compare hashes
+          idsHash' `shouldBe` idsHash
           serviceId5 `shouldBe` serviceId
           Resp "" serviceId6 (ENDS 2) <- tGet1 nh2
           serviceId6 `shouldBe` serviceId