diff --git a/src/Simplex/Messaging/Server.hs b/src/Simplex/Messaging/Server.hs index 813b52fe1..0a035f622 100644 --- a/src/Simplex/Messaging/Server.hs +++ b/src/Simplex/Messaging/Server.hs @@ -1415,7 +1415,7 @@ client let clntServiceId = (\THClientService {serviceId} -> serviceId) <$> (peerClientService =<< thAuth thParams') process batchSubs t acc@(rs, msgs) = (maybe acc (\(!r, !msg_) -> (r : rs, maybe msgs (: msgs) msg_))) - <$> processCommand clntServiceId batchSubs t + <$> processCommand (thVersion thParams') (forkCmd serverClientConcurrency) clntServiceId batchSubs t forever $ do batch <- atomically (readTBQueue rcvQ) processedAt <- liftIO getSystemTime @@ -1572,8 +1572,8 @@ client mkIncProxyStats ps psOwn own sel = do incStat $ sel ps when own $ incStat $ sel psOwn - processCommand :: Maybe ServiceId -> Either ErrorType (Map RecipientId Message, Map RecipientId (Either ErrorType ()), Map RecipientId (Either ErrorType ())) -> VerifiedTransmission s -> M s (Maybe ResponseAndMessage) - processCommand clntServiceId batchSubs (q_, (corrId, entId, cmd)) = case cmd of + processCommand :: VersionSMP -> (CorrId -> EntityId -> BrokerMsg -> M s BrokerMsg -> M s (Maybe ResponseAndMessage)) -> Maybe ServiceId -> Either ErrorType (Map RecipientId Message, Map RecipientId (Either ErrorType ()), Map RecipientId (Either ErrorType ())) -> VerifiedTransmission s -> M s (Maybe ResponseAndMessage) + processCommand clntVersion forkClntCmd clntServiceId batchSubs (q_, (corrId, entId, cmd)) = case cmd of Cmd SProxiedClient command -> processProxiedCmd (corrId, entId, command) Cmd SSender command -> case command of SKEY k -> withQueue $ \q qr -> checkMode QMMessaging qr $ secureQueue_ q k @@ -1582,7 +1582,7 @@ client Cmd SProxyService (RFWD encBlock) -> (response . (corrId, NoEntity,) =<<) <$> processForwardedCommand encBlock Cmd SResolver (RSLV d) -> rslvNamesEnv >>= \case Nothing -> pure $ response (corrId, NoEntity, ERR (NAME NO_RESOLVER)) - Just nenv -> forkCmd serverClientConcurrency corrId NoEntity rslvTooManyRequests (resolveNameMsg (thVersion thParams') nenv d) + Just nenv -> forkClntCmd corrId NoEntity rslvTooManyRequests (resolveNameMsg clntVersion nenv d) Cmd SSenderLink command -> case command of LKEY k -> withQueue $ \q qr -> checkMode QMMessaging qr $ secureQueue_ q k $>> getQueueLink_ q qr LGET -> withQueue $ \q qr -> checkContact qr $ getQueueLink_ q qr @@ -2211,22 +2211,17 @@ client r2 <- liftEitherWith (const BLOCK) $ EncResponse <$> C.cbEncrypt clientSecret (fromMaybe (C.reverseNonce clientNonce) nonce_) r' paddedProxiedTLength let fr = FwdResponse {fwdCorrId, fwdResponse = r2} pure $ RRES nonce_ $ EncFwdResponse $ C.cbEncryptNoPad sessSecret (C.reverseNonce proxyNonce) (smpEncode fr) + forkFwdCmd corrId' entId' timeoutErr cmdAction = forkCmd_ serverResolverConcurrency forwardedCmdWait (sendResp timeoutErr) $ cmdAction >>= sendResp + where + sendResp msg = do + rres <- either ERR id <$> runExceptT (encodeResp (corrId', entId', msg)) + atomically $ writeTBQueue sndQ ([(corrId, NoEntity, rres)], []) -- the inner response, or Nothing if forked (RSLV). r_ <- lift (rejectOrVerify clntThAuth t') >>= \case -- rejectOrVerify filters allowed commands, no need to repeat it here. Left r -> pure $ Just r - Right t''@(_, (corrId', entId', cmd')) -> case cmd' of - Cmd SResolver (RSLV d) -> lift $ rslvNamesEnv >>= \case - Nothing -> pure $ Just (corrId', entId', ERR (NAME NO_RESOLVER)) - Just nenv -> do - let sendResolved msg = do - rres <- either ERR id <$> runExceptT (encodeResp (corrId', entId', msg)) - atomically $ writeTBQueue sndQ ([(corrId, NoEntity, rres)], []) - forkCmd_ serverResolverConcurrency forwardedCmdWait (sendResolved rslvTooManyRequests) $ - resolveNameMsg (thVersion clntTHParams) nenv d >>= sendResolved - -- INTERNAL because processCommand never returns Nothing for sender commands; - -- `fst` drops the empty message only returned for SUB. - _ -> Just . maybe (corrId', entId', ERR INTERNAL) fst <$> lift (processCommand Nothing (Right (M.empty, M.empty, M.empty)) t'') + -- `fst` drops the empty message only returned for SUB. + Right t'' -> fst <$$> lift (processCommand (thVersion clntTHParams) forkFwdCmd Nothing (Right (M.empty, M.empty, M.empty)) t'') stats <- asks serverStats incStat $ pMsgFwdsRecv stats traverse encodeResp r_ >>= \case diff --git a/tests/RSLVTests.hs b/tests/RSLVTests.hs index 0d096bccb..87836a52d 100644 --- a/tests/RSLVTests.hs +++ b/tests/RSLVTests.hs @@ -27,6 +27,7 @@ import NamesResolverServer (memCfg, memCfg2, memProxyCfg, withNames) import qualified NamesResolverServer as NRS import SMPClient import Simplex.Messaging.Client +import Simplex.Messaging.Client.Agent (SMPClientAgentConfig (..)) import qualified Simplex.Messaging.Crypto as C import Simplex.Messaging.Encoding.String (strDecode) import SMPNamesTests (availableBody, registeredBody, reservedBody, resolved, testNameRecord, testPricing) @@ -55,6 +56,7 @@ import Simplex.Messaging.SimplexName (SimplexDomain) import Simplex.Messaging.Transport import Simplex.Messaging.Version (mkVersionRange) import Test.Hspec hiding (fit, it) +import UnliftIO.Async (replicateConcurrently) import Util (HasTestEnv, it) domain :: Text -> SimplexDomain @@ -93,6 +95,8 @@ rslvTests = do describe "RSLV forwarded (PFWD)" $ do it "PFWD-wrapped RSLV reaches resolver via proxy (PCEProtocolError (NAME RESOLVER))" testRslvForwarded it "PFWD-wrapped RSLV success returns RNAME (record JSON frames over the proxy)" testRslvForwardedSuccess + it "PFWD-wrapped RSLV without names config -> NAME NO_RESOLVER" testRslvForwardedDisabled + it "PFWD-wrapped RSLV waiting for a slot fails at the relay wait timeout" testRslvForwardedSlotWaitTimeout describe "RSLV success path (RNAME response)" $ do it "returns RNAME with NameRecord" testRslvSuccess describe "RSLV availability (RNAME response)" $ do @@ -161,6 +165,34 @@ forwardedResolveAlice = do sess <- runExceptT' (connectSMPProxiedRelay pc NRMInteractive relayServ Nothing) runExceptT (proxyResolveName pc NRMInteractive sess (domain "alice.simplex")) +testRslvForwardedDisabled :: HasTestEnv => IO () +testRslvForwardedDisabled = + withSmpServerConfigOn (transport @TLS) memProxyCfg testPort $ \_ -> + withSmpServerConfigOn (transport @TLS) memCfg2 testPort2 $ \_ -> + forwardedResolveAlice `shouldReturn` Left (PCEProtocolError (SMP.NAME SMP.NO_RESOLVER)) + +testRslvForwardedSlotWaitTimeout :: HasTestEnv => IO () +testRslvForwardedSlotWaitTimeout = + NRS.withResolverServerDelayed 3000 (NRS.resolveResp status200 "{}") $ \port _ -> + withSmpServerConfigOn (transport @TLS) memProxyCfg testPort $ \_ -> + withSmpServerConfigOn (transport @TLS) (relayCfg port) testPort2 $ \_ -> do + rs <- replicateConcurrently 2 forwardedResolveAlice + rs `shouldMatchList` [Left (PCEProtocolError (SMP.NAME (SMP.RESOLVER "timeout"))), Left (PCEProtocolError (SMP.NAME (SMP.RESOLVER "too many requests")))] + threadDelay 1500000 + SlotWaitMetrics {waits, waitTimeouts, waitMaxMs} <- readSlotWaitMetrics + (waits, waitTimeouts) `shouldBe` (1, 1) + waitMaxMs `shouldSatisfy` (\t -> t >= 300 && t < 1000) + where + relayCfg port = updateCfg (withNames port memCfg2) $ \c -> + let aCfg = smpAgentCfg c + cCfg = smpCfg aCfg + nt = NetworkTimeout {backgroundTimeout = 300000, interactiveTimeout = 300000} + in c + { serverResolverConcurrency = 1, + prometheusInterval = Just 1, + smpAgentCfg = aCfg {smpCfg = cCfg {networkConfig = (networkConfig cCfg) {tcpTimeout = nt}}} + } + testRslvForwarded :: HasTestEnv => IO () testRslvForwarded = withProxyAndResolver (status404, "{}") $