From b9a07b543c15dd71e78e2a6edac611259597df26 Mon Sep 17 00:00:00 2001 From: Evgeny Poberezkin Date: Sun, 31 Aug 2025 10:59:28 +0100 Subject: [PATCH] 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 | 10 +-- src/Simplex/Messaging/Agent/Client.hs | 9 +- .../Messaging/Agent/NtfSubSupervisor.hs | 2 +- .../Messaging/Agent/Store/AgentStore.hs | 7 +- .../Migrations/M20250815_service_certs.hs | 3 +- .../Store/SQLite/Migrations/agent_schema.sql | 3 +- src/Simplex/Messaging/Client.hs | 10 +-- src/Simplex/Messaging/Client/Agent.hs | 57 ++++++------- src/Simplex/Messaging/Crypto.hs | 6 +- .../Messaging/Notifications/Protocol.hs | 14 +--- src/Simplex/Messaging/Notifications/Server.hs | 19 +++-- .../Messaging/Notifications/Server/Stats.hs | 1 + .../Notifications/Server/Store/Migrations.hs | 76 +++++++++++------ .../Notifications/Server/Store/Postgres.hs | 35 +++----- .../Server/Store/ntf_server_schema.sql | 55 +++++++++++-- src/Simplex/Messaging/Protocol.hs | 74 ++++++++++++++--- src/Simplex/Messaging/Server.hs | 47 ++++++----- .../Messaging/Server/MsgStore/Journal.hs | 4 +- src/Simplex/Messaging/Server/Prometheus.hs | 1 + .../Messaging/Server/QueueStore/Postgres.hs | 18 ++-- .../Server/QueueStore/Postgres/Migrations.hs | 73 ++++++++++------- .../QueueStore/Postgres/server_schema.sql | 53 +++++++----- .../Messaging/Server/QueueStore/STM.hs | 44 ++++++---- .../Messaging/Server/QueueStore/Types.hs | 2 +- src/Simplex/Messaging/Server/Stats.hs | 82 +++++++++++++++++-- .../Messaging/Server/StoreLog/ReadWrite.hs | 2 +- tests/ServerTests.hs | 20 +++-- 27 files changed, 479 insertions(+), 248 deletions(-) diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index 04872ce64..d7270efeb 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/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))) diff --git a/src/Simplex/Messaging/Agent/Client.hs b/src/Simplex/Messaging/Agent/Client.hs index 77e34dcdb..6c0e92c5d 100644 --- a/src/Simplex/Messaging/Agent/Client.hs +++ b/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) diff --git a/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs b/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs index 85fa45c49..262eb7b9f 100644 --- a/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs +++ b/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 diff --git a/src/Simplex/Messaging/Agent/Store/AgentStore.hs b/src/Simplex/Messaging/Agent/Store/AgentStore.hs index 1be6470bc..41a65b04f 100644 --- a/src/Simplex/Messaging/Agent/Store/AgentStore.hs +++ b/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 = diff --git a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20250815_service_certs.hs b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20250815_service_certs.hs index fdbfc9e25..fb76d0f26 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20250815_service_certs.hs +++ b/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 ); diff --git a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql index 5338c7d04..25063a33e 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql +++ b/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); diff --git a/src/Simplex/Messaging/Client.hs b/src/Simplex/Messaging/Client.hs index 5a3118ca1..15cd039fe 100644 --- a/src/Simplex/Messaging/Client.hs +++ b/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 diff --git a/src/Simplex/Messaging/Client/Agent.hs b/src/Simplex/Messaging/Client/Agent.hs index 2b4ef14e1..3756df42d 100644 --- a/src/Simplex/Messaging/Client/Agent.hs +++ b/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) diff --git a/src/Simplex/Messaging/Crypto.hs b/src/Simplex/Messaging/Crypto.hs index fdd8bd0d2..74e2f1fd6 100644 --- a/src/Simplex/Messaging/Crypto.hs +++ b/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. diff --git a/src/Simplex/Messaging/Notifications/Protocol.hs b/src/Simplex/Messaging/Notifications/Protocol.hs index 0b5889bb7..7acb714c0 100644 --- a/src/Simplex/Messaging/Notifications/Protocol.hs +++ b/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 diff --git a/src/Simplex/Messaging/Notifications/Server.hs b/src/Simplex/Messaging/Notifications/Server.hs index ac274dc08..b95f1588f 100644 --- a/src/Simplex/Messaging/Notifications/Server.hs +++ b/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 diff --git a/src/Simplex/Messaging/Notifications/Server/Stats.hs b/src/Simplex/Messaging/Notifications/Server/Stats.hs index a20e41c34..7125ce290 100644 --- a/src/Simplex/Messaging/Notifications/Server/Stats.hs +++ b/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, diff --git a/src/Simplex/Messaging/Notifications/Server/Store/Migrations.hs b/src/Simplex/Messaging/Notifications/Server/Store/Migrations.hs index 155969bbe..1051cc4fb 100644 --- a/src/Simplex/Messaging/Notifications/Server/Store/Migrations.hs +++ b/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 diff --git a/src/Simplex/Messaging/Notifications/Server/Store/Postgres.hs b/src/Simplex/Messaging/Notifications/Server/Store/Postgres.hs index b6f23047f..7f10fc59f 100644 --- a/src/Simplex/Messaging/Notifications/Server/Store/Postgres.hs +++ b/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. diff --git a/src/Simplex/Messaging/Notifications/Server/Store/ntf_server_schema.sql b/src/Simplex/Messaging/Notifications/Server/Store/ntf_server_schema.sql index 030792905..b73995684 100644 --- a/src/Simplex/Messaging/Notifications/Server/Store/ntf_server_schema.sql +++ b/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 ); diff --git a/src/Simplex/Messaging/Protocol.hs b/src/Simplex/Messaging/Protocol.hs index 569fd1665..23f270fc9 100644 --- a/src/Simplex/Messaging/Protocol.hs +++ b/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 diff --git a/src/Simplex/Messaging/Server.hs b/src/Simplex/Messaging/Server.hs index 227b3956e..5b7b72b81 100644 --- a/src/Simplex/Messaging/Server.hs +++ b/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 = diff --git a/src/Simplex/Messaging/Server/MsgStore/Journal.hs b/src/Simplex/Messaging/Server/MsgStore/Journal.hs index 8561e8f3e..d3432ff9a 100644 --- a/src/Simplex/Messaging/Server/MsgStore/Journal.hs +++ b/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 diff --git a/src/Simplex/Messaging/Server/Prometheus.hs b/src/Simplex/Messaging/Server/Prometheus.hs index 859587b60..e4d6a2774 100644 --- a/src/Simplex/Messaging/Server/Prometheus.hs +++ b/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, diff --git a/src/Simplex/Messaging/Server/QueueStore/Postgres.hs b/src/Simplex/Messaging/Server/QueueStore/Postgres.hs index 95e640fe6..a7d7cf305 100644 --- a/src/Simplex/Messaging/Server/QueueStore/Postgres.hs +++ b/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 diff --git a/src/Simplex/Messaging/Server/QueueStore/Postgres/Migrations.hs b/src/Simplex/Messaging/Server/QueueStore/Postgres/Migrations.hs index 0a4fec8c3..2401279ee 100644 --- a/src/Simplex/Messaging/Server/QueueStore/Postgres/Migrations.hs +++ b/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 diff --git a/src/Simplex/Messaging/Server/QueueStore/Postgres/server_schema.sql b/src/Simplex/Messaging/Server/QueueStore/Postgres/server_schema.sql index 76f9ded2f..ff2d25af0 100644 --- a/src/Simplex/Messaging/Server/QueueStore/Postgres/server_schema.sql +++ b/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 ); diff --git a/src/Simplex/Messaging/Server/QueueStore/STM.hs b/src/Simplex/Messaging/Server/QueueStore/STM.hs index 450d99305..e8c174982 100644 --- a/src/Simplex/Messaging/Server/QueueStore/STM.hs +++ b/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 diff --git a/src/Simplex/Messaging/Server/QueueStore/Types.hs b/src/Simplex/Messaging/Server/QueueStore/Types.hs index 55be4d21d..c97e467a7 100644 --- a/src/Simplex/Messaging/Server/QueueStore/Types.hs +++ b/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, diff --git a/src/Simplex/Messaging/Server/Stats.hs b/src/Simplex/Messaging/Server/Stats.hs index 4af195295..c32e8e60b 100644 --- a/src/Simplex/Messaging/Server/Stats.hs +++ b/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 diff --git a/src/Simplex/Messaging/Server/StoreLog/ReadWrite.hs b/src/Simplex/Messaging/Server/StoreLog/ReadWrite.hs index ea6c9ed4a..2fd4ca6d8 100644 --- a/src/Simplex/Messaging/Server/StoreLog/ReadWrite.hs +++ b/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 () diff --git a/tests/ServerTests.hs b/tests/ServerTests.hs index cad598a0e..f34969e72 100644 --- a/tests/ServerTests.hs +++ b/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