diff --git a/src/Simplex/Messaging/Server.hs b/src/Simplex/Messaging/Server.hs index 30133885d..56e83d79d 100644 --- a/src/Simplex/Messaging/Server.hs +++ b/src/Simplex/Messaging/Server.hs @@ -745,12 +745,10 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt _ -> pure () getClientService :: s -> TVar ChaChaDRG -> Int -> SMPServiceRole -> X.CertificateChain -> XV.Fingerprint -> ExceptT TransportError IO ServiceId - getClientService ms g idSize role cert fp = do + getClientService ms g idSize role _ (XV.Fingerprint fp) = do newServiceId <- EntityId <$> atomically (C.randomBytes idSize g) - ts <- liftIO getSystemDate - let sr = ServiceRec {serviceId = newServiceId, serviceRole = role, serviceCert = cert, serviceCertHash = fp, serviceCreatedAt = ts} withExceptT (const $ TEHandshake BAD_SERVICE) $ ExceptT $ - getCreateService @(StoreQueue s) (queueStore ms) sr + fromMaybe newServiceId <$$> getServiceId @(StoreQueue s) (queueStore ms) role fp controlPortThread_ :: ServerConfig s -> [M s ()] controlPortThread_ ServerConfig {controlPort = Just port} = [runCPServer port] @@ -1076,7 +1074,7 @@ runClientTransport h@THandle {params = thParams@THandleParams {sessionId}} = do expCfg <- asks $ inactiveClientExpiration . config th <- newMVar h -- put TH under a fair lock to interleave messages and command responses labelMyThread . B.unpack $ "client $" <> encode sessionId - raceAny_ $ [liftIO $ send th c, liftIO $ sendMsg th c, client s ms c, receive h ms c] <> disconnectThread_ c s expCfg + raceAny_ $ [liftIO $ send th c, liftIO $ sendMsg th c, client s ms (getPeerCertChain $ connection h) c, receive h ms c] <> disconnectThread_ c s expCfg disconnectThread_ :: Client s -> Server s -> Maybe ExpirationConfig -> [M s ()] disconnectThread_ c s (Just expCfg) = [liftIO $ disconnectTransport h (rcvActiveAt c) (sndActiveAt c) expCfg (noSubscriptions c s)] disconnectThread_ _ _ _ = [] @@ -1405,15 +1403,16 @@ forkClient Client {endThreads, endThreadSeq} label action = do action `finally` atomically (modifyTVar' endThreads $ IM.delete tId) mkWeakThreadId t >>= atomically . modifyTVar' endThreads . IM.insert tId -client :: forall s. MsgStoreClass s => Server s -> s -> Client s -> M s () +client :: forall s. MsgStoreClass s => Server s -> s -> X.CertificateChain -> Client s -> M s () client Server {subscribers, ntfSubscribers} ms + serviceCert clnt@Client {clientId, rcvQ, sndQ, msgQ, clientTHParams = thParams'@THandleParams {sessionId}, procThreads} = do labelMyThread . B.unpack $ "client $" <> encode sessionId <> " commands" stats <- asks serverStats authDelay <- asks $ authErrorDelay . config - let clntServiceId = (\THClientService {serviceId} -> serviceId) <$> (peerClientService =<< thAuth thParams') + let clntServiceId = (\THClientService {serviceId} -> serviceId) <$> clntService process batchSubs t acc@(rs, msgs) = (maybe acc (\(!r, !msg_) -> (r : rs, maybe msgs (: msgs) msg_))) <$> processCommand clntServiceId batchSubs t @@ -1424,6 +1423,7 @@ client foldrM (process batchSubs) ([], []) batch >>= \(rs_, msgs) -> mapM_ (liftIO . sendResponses clnt stats authDelay processedAt . (,msgs)) (L.nonEmpty rs_) where + clntService = peerClientService =<< thAuth thParams' prepareBatchSubs :: Maybe ServiceId -> NonEmpty (VerifiedTransmission s) -> @@ -1445,7 +1445,13 @@ client where ifNotNull qs f = if null qs then pure M.empty else f qs setService :: (PartyI p, ServiceParty p) => SParty p -> Maybe ServiceId -> [StoreQueue s] -> ExceptT ErrorType IO (Map RecipientId (Either ErrorType ())) - setService party sId = ExceptT . setQueueServices (queueStore ms) party sId + setService party sId qs = forM_ clntService storeService >> ExceptT (setQueueServices (queueStore ms) party sId qs) + + storeService :: THPeerClientService -> ExceptT ErrorType IO () + storeService THClientService {serviceId, serviceRole, serviceCertHash} = do + serviceCreatedAt <- liftIO getSystemDate + serviceId' <- ExceptT $ getCreateService @(StoreQueue s) (queueStore ms) ServiceRec {serviceId, serviceRole, serviceCert, serviceCertHash, serviceCreatedAt} + unless (serviceId' == serviceId) $ throwE SERVICE processProxiedCmd :: Transmission (Command 'ProxiedClient) -> M s (Maybe ResponseAndMessage) processProxiedCmd (corrId, EntityId sessId, command) = (\t -> ((corrId, EntityId sessId, t), Nothing)) <$$> case command of @@ -1673,7 +1679,7 @@ client updatedAt, rcvServiceId = clntServiceId } - liftIO (addQueue ms rcvId qr) >>= \case + liftIO (runExceptT (forM_ clntService storeService) $>> addQueue ms rcvId qr) >>= \case Left DUPLICATE_ -- TODO [short links] possibly, we somehow need to understand which IDs caused collision to retry if it's not client-supplied? | clntIds -> pure $ ERR AUTH -- no retry on collision if sender ID is client-supplied | otherwise -> tryCreate (n - 1) diff --git a/src/Simplex/Messaging/Server/QueueStore/Postgres.hs b/src/Simplex/Messaging/Server/QueueStore/Postgres.hs index 66e9a91f3..0ce465a32 100644 --- a/src/Simplex/Messaging/Server/QueueStore/Postgres.hs +++ b/src/Simplex/Messaging/Server/QueueStore/Postgres.hs @@ -339,6 +339,15 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where rId = recipientId sq qr = queueRec sq + getServiceId :: PostgresQueueStore q -> SMPServiceRole -> CertFingerprint -> IO (Either ErrorType (Maybe ServiceId)) + getServiceId st serviceRole fp = + E.uninterruptibleMask_ $ runExceptT $ withDB "getServiceId" st $ \db -> + traverse checkRole <$> maybeFirstRow id (DB.query db "SELECT service_id, service_role FROM services WHERE service_cert_hash = ?" (Only (Binary fp))) + where + checkRole (serviceId, role) + | role == serviceRole = Right serviceId + | otherwise = Left SERVICE + getCreateService :: PostgresQueueStore q -> ServiceRec -> IO (Either ErrorType ServiceId) getCreateService st sr@ServiceRec {serviceId = newSrvId, serviceRole, serviceCertHash = XV.Fingerprint fp} = withLockMap (serviceLocks st) fp "getCreateService" $ E.uninterruptibleMask_ $ runExceptT $ do diff --git a/src/Simplex/Messaging/Server/QueueStore/STM.hs b/src/Simplex/Messaging/Server/QueueStore/STM.hs index 8d56897b5..325683f00 100644 --- a/src/Simplex/Messaging/Server/QueueStore/STM.hs +++ b/src/Simplex/Messaging/Server/QueueStore/STM.hs @@ -286,6 +286,14 @@ instance StoreQueueClass q => QueueStoreClass q (STMQueueStore q) where mapM_ (removeNotifier st) $ notifier q pure q + getServiceId :: STMQueueStore q -> SMPServiceRole -> CertFingerprint -> IO (Either ErrorType (Maybe ServiceId)) + getServiceId st serviceRole fp = + traverse checkRole <$> (TM.lookupIO fp (serviceCerts st) $>>= (`TM.lookupIO` services st)) + where + checkRole STMService {serviceRec = ServiceRec {serviceId, serviceRole = role}} + | role == serviceRole = Right serviceId + | otherwise = Left SERVICE + getCreateService :: STMQueueStore q -> ServiceRec -> IO (Either ErrorType ServiceId) getCreateService st sr@ServiceRec {serviceId = newSrvId, serviceRole, serviceCertHash = XV.Fingerprint fp} = TM.lookupIO fp serviceCerts @@ -363,7 +371,7 @@ instance StoreQueueClass q => QueueStoreClass q (STMQueueStore q) where 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 . first (fromIntegral . S.size)) . readTVarIO . serviceSel) + maybe (pure $ Right (0, mempty)) (fmap (Right . first (fromIntegral . S.size)) . readTVarIO . serviceSel) where serviceSel :: STMService -> TVar (Set QueueId, IdsHash) serviceSel = case party of diff --git a/src/Simplex/Messaging/Server/QueueStore/Types.hs b/src/Simplex/Messaging/Server/QueueStore/Types.hs index e8d47751b..b1f813915 100644 --- a/src/Simplex/Messaging/Server/QueueStore/Types.hs +++ b/src/Simplex/Messaging/Server/QueueStore/Types.hs @@ -17,6 +17,7 @@ import Data.Map.Strict (Map) import Simplex.Messaging.Protocol import Simplex.Messaging.Server.QueueStore import Simplex.Messaging.SystemTime +import Simplex.Messaging.Transport (SMPServiceRole) class StoreQueueClass q where recipientId :: q -> RecipientId @@ -43,6 +44,7 @@ class StoreQueueClass q => QueueStoreClass q s where unblockQueue :: s -> q -> IO (Either ErrorType ()) updateQueueTime :: s -> q -> SystemDate -> IO (Either ErrorType QueueRec) deleteStoreQueue :: s -> q -> IO (Either ErrorType QueueRec) + getServiceId :: s -> SMPServiceRole -> CertFingerprint -> IO (Either ErrorType (Maybe ServiceId)) getCreateService :: s -> ServiceRec -> IO (Either ErrorType ServiceId) setQueueService :: (PartyI p, ServiceParty p) => s -> q -> SParty p -> Maybe ServiceId -> IO (Either ErrorType ()) setQueueServices :: (PartyI p, ServiceParty p) => s -> SParty p -> Maybe ServiceId -> [q] -> IO (Either ErrorType (Map RecipientId (Either ErrorType ()))) diff --git a/tests/AgentTests/FunctionalAPITests.hs b/tests/AgentTests/FunctionalAPITests.hs index 7ed0d17b8..d1bcc6b80 100644 --- a/tests/AgentTests/FunctionalAPITests.hs +++ b/tests/AgentTests/FunctionalAPITests.hs @@ -4719,14 +4719,10 @@ testServiceIdChangeOnReconnect ps@(_, ASType qs _) = do ("", "", SERVICE_UP _ _) <- nGet service ("", "", UP _ [_]) <- nGet user pure () - -- Phase 3: Normal reconnect - service should subscribe normally ("", "", SERVICE_DOWN _ _) <- nGet service ("", "", DOWN _ [_]) <- nGet user withSmpServerStoreLogOn ps testPort $ \_ -> do - liftIO $ getInAnyOrder service - [ \case ("", "", AEvt SAENone (SERVICE_UP _ _)) -> True; _ -> False, - \case ("", "", AEvt SAENone (SERVICE_ALL _)) -> True; _ -> False - ] + ("", "", SERVICE_UP _ (SMP.ServiceSubResult (Just SMP.SSErrorServiceId {}) (SMP.ServiceSub _ 0 _))) <- nGet service ("", "", UP _ [_]) <- nGet user pure () diff --git a/tests/ServerTests.hs b/tests/ServerTests.hs index 146a1d46f..a33c155c7 100644 --- a/tests/ServerTests.hs +++ b/tests/ServerTests.hs @@ -53,6 +53,7 @@ import Simplex.Messaging.Server.Stats (PeriodStatsData (..), ServerStatsData (.. import Simplex.Messaging.Server.StoreLog (StoreLogRecord (..)) import Simplex.Messaging.Transport import Simplex.Messaging.Transport.Credentials +import Simplex.Messaging.Transport.Server (loadServerCredential) import System.Directory (doesFileExist, removeFile) import System.IO (IOMode (..), withFile) import System.Timeout @@ -91,6 +92,7 @@ serverTests = do testServiceDeliverSubscribe testServiceUpgradeAndDowngrade testServiceSubsTotalCount + testServiceStoredWithQueue describe "Store log" testWithStoreLog describe "Restore messages" testRestoreMessages describe "Restore messages (old / v2)" testRestoreExpireMessages @@ -935,9 +937,35 @@ testServiceSubsTotalCount = threadDelay 1500000 B.readFile testPrometheusMetricsFile >>= \m -> readServiceSubsMetric (B.unpack m) `shouldBe` Just 3 +testServiceStoredWithQueue :: SpecWith (ASrvTransport, AStoreType) +testServiceStoredWithQueue = + it "should store service with its first queue" $ \(at@(ATransport t), msType) -> do + g <- C.newRandom + tlsCred <- loadServerCredential ntfTestServerCredentials + serviceKeys@(_, servicePK) <- atomically $ C.generateKeyPair g + (rPub, rKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g + (dhPub, _ :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g + let cfg' = updateCfg (cfgMS msType) $ \cfg_ -> cfg_ {prometheusInterval = Just 1} + servicesCount = do + threadDelay 1500000 + readMetric "simplex_smp_rcv_services_count " . B.unpack <$> B.readFile testPrometheusMetricsFile + withSmpServerConfigOn at cfg' testPort $ \_ -> runSMPClient t $ \h -> do + Resp "1" NoEntity (Ids rId _ _) <- signSendRecv h rKey ("1", NoEntity, New rPub dhPub) + runSMPServiceClient t (tlsCred, serviceKeys) $ \sh -> do + servicesCount `shouldReturn` Just 0 + runSMPServiceClient t (tlsCred, serviceKeys) $ \sh' -> do + Resp "2" _ (SOK (Just _)) <- serviceSignSendRecv sh' rKey servicePK ("2", rId, SUB) + servicesCount `shouldReturn` Just 1 + Resp "3" _ (ERR SERVICE) <- serviceSignSendRecv sh rKey servicePK ("3", rId, SUB) + pure () + testNtfServiceClient t serviceKeys (\_ -> pure ()) `shouldThrow` anyErrorCall + readServiceSubsMetric :: String -> Maybe Int -readServiceSubsMetric content = - case filter ("simplex_smp_subscribtion_service_subs_total " `isPrefixOf`) (lines content) of +readServiceSubsMetric = readMetric "simplex_smp_subscribtion_service_subs_total " + +readMetric :: String -> String -> Maybe Int +readMetric metric content = + case filter (metric `isPrefixOf`) (lines content) of (line : _) -> case words line of [_, val, _] -> readMaybe val [_, val] -> readMaybe val