diff --git a/simplexmq.cabal b/simplexmq.cabal index 15bad9c3e..f9a1731a5 100644 --- a/simplexmq.cabal +++ b/simplexmq.cabal @@ -264,7 +264,6 @@ library Simplex.Messaging.Server.Names Simplex.Messaging.Server.Names.Eth.RPC Simplex.Messaging.Server.Names.Eth.SNRC - Simplex.Messaging.Server.Names.Resolver Simplex.Messaging.Server.NtfStore Simplex.Messaging.Server.Prometheus Simplex.Messaging.Server.QueueStore @@ -365,7 +364,6 @@ library , network-uri >=2.6 && <2.7 , optparse-applicative >=0.15 && <0.17 , process ==1.6.* - , psqueues >=0.2.7 && <0.3 , temporary ==1.3.* , wai >=3.2 && <3.3 , wai-app-static >=3.1 && <3.2 diff --git a/src/Simplex/Messaging/Protocol.hs b/src/Simplex/Messaging/Protocol.hs index 3a5f88903..8da8e66b0 100644 --- a/src/Simplex/Messaging/Protocol.hs +++ b/src/Simplex/Messaging/Protocol.hs @@ -171,9 +171,6 @@ module Simplex.Messaging.Protocol NameLink, mkNameLink, unNameLink, - nameRecBytes, - parseNameRec, - smpListPUpTo, MsgFlags (..), initialSMPClientVersion, currentSMPClientVersion, @@ -249,6 +246,7 @@ 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 +import qualified Data.ByteArray.Encoding as BAE import qualified Data.ByteString.Lazy as LB import Data.Char (isPrint, isSpace) import Data.Constraint (Dict (..)) @@ -750,10 +748,16 @@ unNameOwner :: NameOwner -> ByteString unNameOwner (NameOwner bs) = bs {-# INLINE unNameOwner #-} -instance Encoding NameOwner where - smpEncode (NameOwner bs) = bs - {-# INLINE smpEncode #-} - smpP = NameOwner <$> A.take 20 +instance J.ToJSON NameOwner where + toJSON (NameOwner bs) = J.String $ "0x" <> decodeLatin1 (BAE.convertToBase BAE.Base16 bs) + toEncoding (NameOwner bs) = J.toEncoding $ "0x" <> decodeLatin1 (BAE.convertToBase BAE.Base16 bs) + +instance J.FromJSON NameOwner where + parseJSON = J.withText "NameOwner" $ \t -> do + let hex = maybe t id (T.stripPrefix "0x" t) + case BAE.convertFromBase BAE.Base16 (encodeUtf8 hex) of + Left e -> fail e + Right bs -> either fail pure (mkNameOwner bs) -- | A name-record link (channel or contact). Bare constructor not exported; -- use `mkNameLink` to enforce the ≤1024-byte UTF-8 invariant. @@ -769,20 +773,17 @@ unNameLink :: NameLink -> Text unNameLink (NameLink t) = t {-# INLINE unNameLink #-} -instance Encoding NameLink where - smpEncode (NameLink t) = - let bs = encodeUtf8 t - in smpEncode @Word16 (fromIntegral $ B.length bs) <> bs - smpP = do - n <- fromIntegral <$> smpP @Word16 - when (n > 1024) $ fail "NameLink too long" - bs <- A.take n - either (fail . show) (pure . NameLink) (decodeUtf8' bs) +instance J.ToJSON NameLink where + toJSON (NameLink t) = J.toJSON t + toEncoding (NameLink t) = J.toEncoding t + +instance J.FromJSON NameLink where + parseJSON = J.withText "NameLink" (either fail pure . mkNameLink) -- | Resolved name record returned by the names role. --- Field additions are gated by future SMP version bumps (matching IDS QIK precedent). +-- Wire format is JSON — change requires an SMP version bump. data NameRecord = NameRecord - { nrDisplayName :: Text, -- ≤255 bytes UTF-8 (enforced by Encoding ByteString length prefix) + { nrDisplayName :: Text, nrOwner :: NameOwner, nrChannelLinks :: [NameLink], nrContactLinks :: [NameLink], @@ -793,38 +794,33 @@ data NameRecord = NameRecord } deriving (Eq, Show) --- | Bounded list parser — caps element count before allocating. -smpListPUpTo :: Encoding a => Int -> Parser [a] -smpListPUpTo cap = do - n <- lenP - when (n > cap) $ fail "list too long" - A.count n smpP +instance J.ToJSON NameRecord where + toJSON NameRecord {nrDisplayName, nrOwner, nrChannelLinks, nrContactLinks, nrAdminAddress, nrAdminEmail, nrExpiry, nrIsTest} = + J.object + [ "displayName" J..= nrDisplayName, + "owner" J..= nrOwner, + "channelLinks" J..= nrChannelLinks, + "contactLinks" J..= nrContactLinks, + "adminAddress" J..= nrAdminAddress, + "adminEmail" J..= nrAdminEmail, + "expiry" J..= nrExpiry, + "isTest" J..= nrIsTest + ] --- | Encode NameRecord on the wire. Version-branched in the same shape as IDS QIK. -nameRecBytes :: VersionSMP -> NameRecord -> ByteString -nameRecBytes _v NameRecord {nrDisplayName, nrOwner, nrChannelLinks, nrContactLinks, nrAdminAddress, nrAdminEmail, nrExpiry, nrIsTest} = - smpEncode nrDisplayName - <> smpEncode nrOwner - <> smpEncodeList nrChannelLinks - <> smpEncodeList nrContactLinks - <> smpEncode nrAdminAddress - <> smpEncode nrAdminEmail - <> smpEncode nrExpiry - <> smpEncode nrIsTest - --- | Parse NameRecord. Combined channel+contact list cap is 8. -parseNameRec :: VersionSMP -> Parser NameRecord -parseNameRec _v = do - nrDisplayName <- smpP - nrOwner <- smpP - nrChannelLinks <- smpListPUpTo 8 - nrContactLinks <- smpListPUpTo (8 - length nrChannelLinks) - nrAdminAddress <- smpP - nrAdminEmail <- smpP - nrExpiry <- smpP - when (nrExpiry < 0) $ fail "expiry must be non-negative" - nrIsTest <- smpP - pure NameRecord {nrDisplayName, nrOwner, nrChannelLinks, nrContactLinks, nrAdminAddress, nrAdminEmail, nrExpiry, nrIsTest} +instance J.FromJSON NameRecord where + parseJSON = J.withObject "NameRecord" $ \o -> do + nrDisplayName <- o J..: "displayName" + nrOwner <- o J..: "owner" + nrChannelLinks <- o J..: "channelLinks" + nrContactLinks <- o J..: "contactLinks" + when (length nrChannelLinks + length nrContactLinks > 8) $ + fail "combined channelLinks + contactLinks > 8" + nrAdminAddress <- o J..:? "adminAddress" + nrAdminEmail <- o J..:? "adminEmail" + nrExpiry <- o J..: "expiry" + when (nrExpiry < 0) $ fail "expiry must be non-negative" + nrIsTest <- o J..: "isTest" + pure NameRecord {nrDisplayName, nrOwner, nrChannelLinks, nrContactLinks, nrAdminAddress, nrAdminEmail, nrExpiry, nrIsTest} data BrokerMsg where -- SMP broker messages (responses, client messages, notifications) @@ -2080,7 +2076,7 @@ instance ProtocolEncoding SMPVersion ErrorType BrokerMsg where _ -> err PONG -> e PONG_ NAME rec - | v >= namesSMPVersion -> e (NAME_, ' ') <> nameRecBytes v rec + | v >= namesSMPVersion -> e (NAME_, ' ', Tail (LB.toStrict (J.encode rec))) | otherwise -> e (ERR_, ' ', AUTH) -- pre-v20: shouldn't reach here, degrade to AUTH where e :: Encoding a => a -> ByteString @@ -2130,7 +2126,9 @@ instance ProtocolEncoding SMPVersion ErrorType BrokerMsg where ERR_ -> ERR <$> _smpP PONG_ -> pure PONG NAME_ - | v >= namesSMPVersion -> NAME <$> (A.space *> parseNameRec v) + | v >= namesSMPVersion -> do + Tail bs <- _smpP + either fail (pure . NAME) (J.eitherDecodeStrict bs) | otherwise -> fail "NAME requires namesSMPVersion" where serviceRespP resp diff --git a/src/Simplex/Messaging/Server.hs b/src/Simplex/Messaging/Server.hs index e28b303bf..20e3f2fbf 100644 --- a/src/Simplex/Messaging/Server.hs +++ b/src/Simplex/Messaging/Server.hs @@ -661,8 +661,8 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt map tshow [_pRequests, _pSuccesses, _pErrorsConnect, _pErrorsCompat, _pErrorsOther] showServiceStats ServiceStatsData {_srvAssocNew, _srvAssocDuplicate, _srvAssocUpdated, _srvAssocRemoved, _srvSubCount, _srvSubDuplicate, _srvSubQueues, _srvSubEnd} = map tshow [_srvAssocNew, _srvAssocDuplicate, _srvAssocUpdated, _srvAssocRemoved, _srvSubCount, _srvSubDuplicate, _srvSubQueues, _srvSubEnd] - showNameResolverStats NameResolverStatsData {_rslvReqs, _rslvSucc, _rslvNotFound, _rslvEthErrs, _rslvCacheHits, _rslvCacheMiss, _rslvDisabled} = - map tshow [_rslvReqs, _rslvSucc, _rslvNotFound, _rslvEthErrs, _rslvCacheHits, _rslvCacheMiss, _rslvDisabled] + showNameResolverStats NameResolverStatsData {_rslvReqs, _rslvSucc, _rslvNotFound, _rslvEthErrs, _rslvDisabled} = + map tshow [_rslvReqs, _rslvSucc, _rslvNotFound, _rslvEthErrs, _rslvDisabled] prometheusMetricsThread_ :: ServerConfig s -> [M s ()] prometheusMetricsThread_ ServerConfig {prometheusInterval = Just interval, prometheusMetricsFile} = diff --git a/src/Simplex/Messaging/Server/Env/STM.hs b/src/Simplex/Messaging/Server/Env/STM.hs index 6f6ca1aaf..a23963b1c 100644 --- a/src/Simplex/Messaging/Server/Env/STM.hs +++ b/src/Simplex/Messaging/Server/Env/STM.hs @@ -614,9 +614,8 @@ newEnv config@ServerConfig {allowSMPProxy, smpCredentials, httpCredentials, serv Just nc -> do logInfo $ "[NAMES] resolver enabled, endpoint=" <> scrubUrl (ethereumEndpoint nc) when allowSMPProxy $ - logWarn "[NAMES] enable: on on a proxy-role host: slow RSLV cache misses can serialise other forwarded commands on the same proxy-relay session. For high-volume deployments, run [NAMES] on a separate host." - let rs = rslvStats serverStats - env <- newNamesEnv nc (rslvCacheHits rs) (rslvCacheMiss rs) + logWarn "[NAMES] enable: on on a proxy-role host: slow RSLV calls can serialise other forwarded commands on the same proxy-relay session. For high-volume deployments, run [NAMES] on a separate host." + env <- newNamesEnv nc -- Probe the endpoint at startup. Don't exitFailure: a flapping -- network or an Ethereum host coming up minutes after smp-server -- should not block the server. Log so operators can spot it. diff --git a/src/Simplex/Messaging/Server/Main.hs b/src/Simplex/Messaging/Server/Main.hs index 9dfec8aba..80f09e0da 100644 --- a/src/Simplex/Messaging/Server/Main.hs +++ b/src/Simplex/Messaging/Server/Main.hs @@ -812,9 +812,6 @@ readNamesConfig ini { ethereumEndpoint = either (error . ("[NAMES] ethereum_endpoint: " <>)) id (validateUrl endpoint rpcAuth_), snrcAddress = either (error . ("[NAMES] snrc_address: " <>)) id $ parseEthAddr (requiredText "snrc_address"), rpcAuth = rpcAuth_, - cacheSeconds = readIniDefault 300 "NAMES" "cache_seconds" ini, - cacheMaxEntries = readIniDefault 100000 "NAMES" "cache_max_entries" ini, - cacheMaxBytes = readIniDefault 67108864 "NAMES" "cache_max_bytes" ini, rpcTimeoutMs = readIniDefault 3000 "NAMES" "rpc_timeout_ms" ini, rpcMaxResponseBytes = readIniDefault 262144 "NAMES" "rpc_max_response_bytes" ini, rpcMaxConcurrency = readIniDefault 8 "NAMES" "rpc_max_concurrency" ini diff --git a/src/Simplex/Messaging/Server/Main/Init.hs b/src/Simplex/Messaging/Server/Main/Init.hs index bc6689d85..659845f99 100644 --- a/src/Simplex/Messaging/Server/Main/Init.hs +++ b/src/Simplex/Messaging/Server/Main/Init.hs @@ -168,10 +168,6 @@ iniFileContent cfgPath logPath opts host basicAuth controlPortPwds = \# ethereum_endpoint: https://eth.simplex.chat:443\n\ \# rpc_auth: basic :\n\ \# snrc_address: 0x\n\ - \# (cache_max_entries and cache_max_bytes both cap the cache; whichever fills first triggers FIFO eviction)\n\ - \# cache_seconds: 300\n\ - \# cache_max_entries: 100000\n\ - \# cache_max_bytes: 67108864\n\ \# rpc_timeout_ms: 3000\n\ \# rpc_max_response_bytes: 262144\n\ \# rpc_max_concurrency: 8\n\n\ diff --git a/src/Simplex/Messaging/Server/Names.hs b/src/Simplex/Messaging/Server/Names.hs index a3088a79d..640618317 100644 --- a/src/Simplex/Messaging/Server/Names.hs +++ b/src/Simplex/Messaging/Server/Names.hs @@ -1,19 +1,130 @@ --- | SMP public-namespace resolver façade. +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE StrictData #-} + +-- | Public-namespace resolver. Each RSLV becomes one eth_call to the +-- configured Ethereum endpoint, bounded by rpcMaxConcurrency and +-- rpcTimeoutMs. Zero-owner / expired records map to NotFound. -- --- Re-exports the resolver's public surface from Names.Resolver and the --- HTTP auth type from Names.Eth.RPC. Implementation lives in Resolver.hs; --- Eth.RPC / Eth.SNRC are transport / codec internals. +-- Transport details live in Names.Eth.RPC (HTTP + JSON-RPC + auth); +-- Keccak-256 namehash and SNRC ABI decoder live in Names.Eth.SNRC. module Simplex.Messaging.Server.Names ( NamesConfig (..), RpcAuth (..), - NamesEnv, + NamesEnv (..), + EthCall, ResolveError (..), newNamesEnv, + newNamesEnvWith, closeNamesEnv, pingEndpoint, resolveName, ) where -import Simplex.Messaging.Server.Names.Eth.RPC (RpcAuth (..)) -import Simplex.Messaging.Server.Names.Resolver (NamesConfig (..), NamesEnv, ResolveError (..), closeNamesEnv, newNamesEnv, pingEndpoint, resolveName) +import qualified Control.Exception as E +import Control.Logger.Simple (logError) +import Data.ByteString.Char8 (ByteString) +import Data.Maybe (fromMaybe) +import Data.Text (Text) +import qualified Data.Text as T +import Data.Time.Clock.POSIX (getPOSIXTime) +import Simplex.Messaging.Protocol (NameOwner, NameRecord (..), unNameOwner) +import Simplex.Messaging.Server.Names.Eth.RPC (EthRpcEnv, EthRpcError (..), RpcAuth (..), closeEthRpcEnv, ethCallReal, newEthRpcEnv) +import Simplex.Messaging.Server.Names.Eth.SNRC (decodeGetRecord, encodeGetRecord, namehash) +import System.Timeout (timeout) + +data NamesConfig = NamesConfig + { ethereumEndpoint :: Text, + snrcAddress :: NameOwner, + rpcAuth :: Maybe RpcAuth, + rpcTimeoutMs :: Int, + rpcMaxResponseBytes :: Int, + rpcMaxConcurrency :: Int + } + deriving (Show) + +data ResolveError + = NotFound + | EthHttpErr + | EthRpcErr {rpcCode :: Int, rpcMessage :: Text} + | EthDecodeErr + | TimedOut + deriving (Eq, Show) + +-- | Test seam: a function from (to, data) -> raw return bytes or error. +-- Production wires this to ethCallReal; tests substitute a stub. +type EthCall = ByteString -> ByteString -> IO (Either EthRpcError ByteString) + +data NamesEnv = NamesEnv + { config :: NamesConfig, + ethCall :: EthCall, + rpcEnv :: Maybe EthRpcEnv -- Nothing for test stubs + } + +newNamesEnv :: NamesConfig -> IO NamesEnv +newNamesEnv cfg = do + rpc <- newEthRpcEnv (ethereumEndpoint cfg) (rpcAuth cfg) (rpcMaxResponseBytes cfg) (rpcMaxConcurrency cfg) + newNamesEnvWith cfg (ethCallReal rpc) (Just rpc) + +-- | Allocate resolver with an injected ethCall (test seam). +newNamesEnvWith :: NamesConfig -> EthCall -> Maybe EthRpcEnv -> IO NamesEnv +newNamesEnvWith config ethCall rpcEnv = pure NamesEnv {config, ethCall, rpcEnv} + +closeNamesEnv :: NamesEnv -> IO () +closeNamesEnv NamesEnv {rpcEnv} = mapM_ closeEthRpcEnv rpcEnv + +-- | Reach the configured endpoint with a harmless probe call to confirm +-- network reachability. Returns Left only on transport-level failures; +-- JSON-RPC errors (misconfigured snrc_address etc.) are treated as +-- "endpoint reachable" — that distinction surfaces later via rslvEthErrs. +pingEndpoint :: NamesEnv -> IO (Either EthRpcError ()) +pingEndpoint NamesEnv {ethCall, config} = + ethCall (unNameOwner (snrcAddress config)) (encodeGetRecord (namehash "")) >>= \case + Left e@(HttpFailure _) -> pure (Left e) + Left e@(HttpStatusErr _) -> pure (Left e) + _ -> pure (Right ()) + +-- | Resolve a lookup key with an rpcTimeoutMs ceiling. Synchronous +-- exceptions are caught and logged; async exceptions propagate. +resolveName :: NamesEnv -> ByteString -> IO (Either ResolveError NameRecord) +resolveName env key = do + r <- E.try (timeout (rpcTimeoutMs (config env) * 1000) (fetch env key)) + case r of + Right result -> pure (fromMaybe (Left TimedOut) result) + Left e + | Just (_ :: E.SomeAsyncException) <- E.fromException e -> E.throwIO e + | otherwise -> do + logError $ "[NAMES] resolver fetch raised " <> T.pack (E.displayException e) + pure (Left EthHttpErr) + +fetch :: NamesEnv -> ByteString -> IO (Either ResolveError NameRecord) +fetch NamesEnv {ethCall, config} key = + ethCall (unNameOwner (snrcAddress config)) (encodeGetRecord (namehash key)) >>= \case + Left e -> pure (Left (mapEthRpcError e)) + Right ret -> case decodeGetRecord ret of + Right Nothing -> pure (Left NotFound) + Right (Just rec) -> checkExpiry rec + Left _ -> pure (Left EthDecodeErr) + where + -- Defense in depth: the SNRC contract should already return the + -- zero-owner sentinel for expired records, but a buggy / pre-upgrade + -- contract might not. nrExpiry == 0 means "never expires" (reserved + -- names); any positive expiry in the past is treated as NotFound. + checkExpiry rec = do + nowSec <- floor <$> getPOSIXTime + pure $ if nrExpiry rec /= 0 && nrExpiry rec < nowSec + then Left NotFound + else Right rec + +-- | Collapse the JSON-RPC transport-layer error space into the resolver's +-- public error space. +mapEthRpcError :: EthRpcError -> ResolveError +mapEthRpcError = \case + HttpFailure _ -> EthHttpErr + HttpStatusErr _ -> EthHttpErr + BodyTooLarge -> EthDecodeErr + InvalidJson _ -> EthDecodeErr + JsonRpcErr c m -> EthRpcErr {rpcCode = c, rpcMessage = m} diff --git a/src/Simplex/Messaging/Server/Names/Resolver.hs b/src/Simplex/Messaging/Server/Names/Resolver.hs deleted file mode 100644 index 1ed654659..000000000 --- a/src/Simplex/Messaging/Server/Names/Resolver.hs +++ /dev/null @@ -1,276 +0,0 @@ -{-# LANGUAGE LambdaCase #-} -{-# LANGUAGE NamedFieldPuns #-} -{-# LANGUAGE OverloadedStrings #-} -{-# LANGUAGE RecordWildCards #-} -{-# LANGUAGE ScopedTypeVariables #-} -{-# LANGUAGE StrictData #-} - --- | Public-namespace resolver: TTL+FIFO cache, in-flight coalescing, --- timeout-bounded RPC, and zero-owner → NotFound mapping. -module Simplex.Messaging.Server.Names.Resolver - ( NamesConfig (..), - RpcAuth (..), - NamesEnv (..), - EthCall, - ResolveError (..), - newNamesEnv, - newNamesEnvWith, - closeNamesEnv, - pingEndpoint, - resolveName, - ) -where - -import Control.Concurrent.STM -import qualified Control.Exception as E -import Control.Logger.Simple (logError) -import Data.ByteString.Char8 (ByteString) -import qualified Data.HashPSQ as PSQ -import Data.IORef (IORef) -import Data.Text (Text) -import qualified Data.Text as T -import Data.Word (Word64) -import Data.Time.Clock.POSIX (getPOSIXTime) -import GHC.Clock (getMonotonicTimeNSec) -import qualified Data.ByteString.Char8 as B -import qualified Data.Text.Encoding as T -import Simplex.Messaging.Protocol (NameLink, NameOwner, NameRecord (..), unNameLink, unNameOwner) -import Simplex.Messaging.Server.Names.Eth.RPC (EthRpcEnv, EthRpcError (..), RpcAuth (..), closeEthRpcEnv, ethCallReal, newEthRpcEnv) -import Simplex.Messaging.Server.Names.Eth.SNRC (decodeGetRecord, encodeGetRecord, namehash) -import Simplex.Messaging.Util (atomicModifyIORef'_) -import System.Timeout (timeout) - --- | Public-namespace resolver configuration. -data NamesConfig = NamesConfig - { ethereumEndpoint :: Text, - snrcAddress :: NameOwner, - rpcAuth :: Maybe RpcAuth, - cacheSeconds :: Int, - cacheMaxEntries :: Int, - cacheMaxBytes :: Int, - rpcTimeoutMs :: Int, - rpcMaxResponseBytes :: Int, - rpcMaxConcurrency :: Int - } - deriving (Show) - -data ResolveError - = NotFound - | EthHttpErr - | EthRpcErr {rpcCode :: Int, rpcMessage :: Text} - | EthDecodeErr - | TimedOut - deriving (Eq, Show) - --- | Test seam: a function from (to, data) -> raw return bytes or error. --- Production wires this to ethCallReal; tests substitute a stub. -type EthCall = ByteString -> ByteString -> IO (Either EthRpcError ByteString) - --- | Cache value bundles a result (NameRecord or NotFound sentinel) with --- its insertion-time byte cost and per-entry TTL (NotFound expires faster --- than positive results so newly-registered names become visible quickly --- while still preventing DoS via unique-name spam). -data CacheEntry = CacheEntry - { ceResult :: Maybe NameRecord, -- Nothing = NotFound; Just = Found - ceBytes :: Int, - ceTtlNs :: Word64 - } - --- | Cache state: (PSQ keyed by LookupKey, priority = insert time in ns, total bytes). --- PSQ minView returns lowest-priority element → FIFO eviction by insertion order. -type CacheState = (PSQ.HashPSQ ByteString Word64 CacheEntry, Int) - -data NamesEnv = NamesEnv - { config :: NamesConfig, - ethCall :: EthCall, - cache :: TVar CacheState, - inflight :: TVar (PSQ.HashPSQ ByteString Word64 (TMVar (Either ResolveError NameRecord))), - rpcEnv :: Maybe EthRpcEnv, -- Nothing for test stubs - cacheHitsRef :: IORef Int, -- shared with ServerStats.rslvStats.rslvCacheHits - cacheMissRef :: IORef Int -- shared with ServerStats.rslvStats.rslvCacheMiss - } - --- | Allocate resolver with real HTTP transport. --- `cacheHitsRef` and `cacheMissRef` are shared with ServerStats.rslvStats so --- the periodic CSV / Prometheus exporter sees per-request cache outcomes. -newNamesEnv :: NamesConfig -> IORef Int -> IORef Int -> IO NamesEnv -newNamesEnv cfg cacheHitsRef cacheMissRef = do - rpc <- newEthRpcEnv (ethereumEndpoint cfg) (rpcAuth cfg) (rpcMaxResponseBytes cfg) (rpcMaxConcurrency cfg) - let call to dat = ethCallReal rpc to dat - newNamesEnvWith cfg call (Just rpc) cacheHitsRef cacheMissRef - --- | Allocate resolver with an injected ethCall (test seam). -newNamesEnvWith :: NamesConfig -> EthCall -> Maybe EthRpcEnv -> IORef Int -> IORef Int -> IO NamesEnv -newNamesEnvWith config ethCall rpcEnv cacheHitsRef cacheMissRef = do - cache <- newTVarIO (PSQ.empty, 0) - inflight <- newTVarIO PSQ.empty - pure NamesEnv {config, ethCall, cache, inflight, rpcEnv, cacheHitsRef, cacheMissRef} - -closeNamesEnv :: NamesEnv -> IO () -closeNamesEnv NamesEnv {rpcEnv} = maybe (pure ()) closeEthRpcEnv rpcEnv - --- | Reach the configured endpoint with a harmless probe call to confirm --- network reachability and basic config sanity. Returns Left only on --- transport-level failures (DNS, TLS, refused) — a JSON-RPC error (e.g. --- a misconfigured snrc_address) is treated as "endpoint reachable", --- because the operator-friendly signal we want is "is the eth host alive, --- not is your contract address right." That distinction surfaces later --- via the rslvEthErrs counter. -pingEndpoint :: NamesEnv -> IO (Either EthRpcError ()) -pingEndpoint NamesEnv {ethCall, config} = do - let to = unNameOwner (snrcAddress config) - -- Use the ENS-style root node (32 zero bytes) — always a valid - -- bytes32 input that costs the contract nothing to look up. - callData = encodeGetRecord (namehash "") - ethCall to callData >>= \case - Left e@(HttpFailure _) -> pure (Left e) - Left e@(HttpStatusErr _) -> pure (Left e) - _ -> pure (Right ()) - --- | Resolve a lookup key. Coalesces concurrent identical requests, caches --- results for cacheSeconds, and bounds RPCs by rpcTimeoutMs. -resolveName :: NamesEnv -> ByteString -> IO (Either ResolveError NameRecord) -resolveName env key = do - now <- getMonotonicTimeNSec - cacheLookup env key now >>= \case - Just result -> do - atomicModifyIORef'_ (cacheHitsRef env) (+ 1) - pure $ maybe (Left NotFound) Right result - Nothing -> do - atomicModifyIORef'_ (cacheMissRef env) (+ 1) - coalesce env key now - --- | Look up the key in cache. Returns: --- Nothing — cache miss (or expired entry, which is evicted) --- Just Nothing — cache hit for NotFound --- Just (Just rec) — cache hit for a NameRecord -cacheLookup :: NamesEnv -> ByteString -> Word64 -> IO (Maybe (Maybe NameRecord)) -cacheLookup NamesEnv {cache} key now = atomically $ do - (psq, totalBytes) <- readTVar cache - case PSQ.lookup key psq of - Just (insertedAt, ce) - | now < insertedAt + ceTtlNs ce -> pure (Just (ceResult ce)) - | otherwise -> do - -- Expired: evict and signal miss. - writeTVar cache (PSQ.delete key psq, totalBytes - ceBytes ce) - pure Nothing - Nothing -> pure Nothing - -ttlFoundNs :: NamesConfig -> Word64 -ttlFoundNs cfg = fromIntegral (cacheSeconds cfg) * 1000000000 - --- | NotFound cache TTL — short enough that a newly-registered name becomes --- visible within seconds, long enough to absorb a unique-name DoS burst. --- Bounded by cacheSeconds in case the operator deliberately ran a tiny TTL. -ttlNotFoundNs :: NamesConfig -> Word64 -ttlNotFoundNs cfg = min (ttlFoundNs cfg) (30 * 1000000000) - --- | Leader/waiter coalescing. Leader runs the RPC under E.mask; waiters --- block on the leader's TMVar. Cleanup runs even on async exception. -coalesce :: NamesEnv -> ByteString -> Word64 -> IO (Either ResolveError NameRecord) -coalesce env@NamesEnv {inflight} key now = do - ticket <- atomically $ do - flight <- readTVar inflight - case PSQ.lookup key flight of - Just (_, mv) -> pure (Right mv) - Nothing -> do - mv <- newEmptyTMVar - writeTVar inflight (PSQ.insert key now mv flight) - pure (Left mv) - case ticket of - Right mv -> atomically (readTMVar mv) -- waiter - Left mv -> E.mask $ \restore -> do - -- Run the fetch with sync-only catching: async exceptions (cancel, - -- killThread) must propagate after we've completed the STM cleanup - -- so waiters never block on an orphan TMVar. - r <- - E.try (restore (fetchOnceTimed env key)) >>= \case - Right ok -> pure ok - Left e - | Just (_ :: E.SomeAsyncException) <- E.fromException e -> do - -- Tell waiters the lookup failed, then rethrow. - atomically $ do - putTMVar mv (Left EthHttpErr) - modifyTVar' inflight (PSQ.delete key) - E.throwIO e - | otherwise -> do - logError $ "[NAMES] resolver fetch raised " <> T.pack (E.displayException e) - pure (Left EthHttpErr) - atomically $ do - putTMVar mv r - modifyTVar' inflight (PSQ.delete key) - case r of - Right rec -> cacheInsert env key now (Just rec) (ttlFoundNs (config env)) - Left NotFound -> cacheInsert env key now Nothing (ttlNotFoundNs (config env)) - Left _ -> pure () -- transient errors (HTTP, decode, timeout) are not cached - pure r - -fetchOnceTimed :: NamesEnv -> ByteString -> IO (Either ResolveError NameRecord) -fetchOnceTimed env key = - timeout (rpcTimeoutMs (config env) * 1000) (fetchOnce env key) >>= \case - Just r -> pure r - Nothing -> pure (Left TimedOut) - -fetchOnce :: NamesEnv -> ByteString -> IO (Either ResolveError NameRecord) -fetchOnce NamesEnv {ethCall, config} key = - ethCall (unNameOwner (snrcAddress config)) (encodeGetRecord (namehash key)) >>= \case - Left e -> pure (Left (mapEthRpcError e)) - Right ret -> case decodeGetRecord ret of - Right Nothing -> pure (Left NotFound) - Right (Just rec) -> checkExpiry rec - Left _ -> pure (Left EthDecodeErr) - where - -- Defense in depth: the SNRC contract should already return the - -- zero-owner sentinel for expired records, but a buggy / pre-upgrade - -- contract might not. nrExpiry == 0 means "never expires" (reserved - -- names); any positive expiry in the past is treated as NotFound. - checkExpiry rec = do - nowSec <- floor <$> getPOSIXTime - pure $ if nrExpiry rec /= 0 && nrExpiry rec < nowSec - then Left NotFound - else Right rec - --- | Collapse the JSON-RPC transport-layer error space into the resolver's --- public error space. Reused by fetchOnce and pingEndpoint. -mapEthRpcError :: EthRpcError -> ResolveError -mapEthRpcError = \case - HttpFailure _ -> EthHttpErr - HttpStatusErr _ -> EthHttpErr - BodyTooLarge -> EthDecodeErr - InvalidJson _ -> EthDecodeErr - JsonRpcErr c m -> EthRpcErr {rpcCode = c, rpcMessage = m} - -cacheInsert :: NamesEnv -> ByteString -> Word64 -> Maybe NameRecord -> Word64 -> IO () -cacheInsert NamesEnv {config, cache} key now result ttl = atomically $ do - (psq, totalBytes) <- readTVar cache - let entryBytes = maybe notFoundOverhead estimateBytes result - (psq', totalBytes') = evictWhile psq totalBytes - evictWhile p tb - | PSQ.size p > cacheMaxEntries config || tb + entryBytes > cacheMaxBytes config = - case PSQ.minView p of - Just (_, _, ce, rest) -> evictWhile rest (tb - ceBytes ce) - Nothing -> (p, tb) - | otherwise = (p, tb) - ce = CacheEntry {ceResult = result, ceBytes = entryBytes, ceTtlNs = ttl} - writeTVar cache (PSQ.insert key now ce psq', totalBytes' + entryBytes) - where - notFoundOverhead = 128 -- PSQ node + key copy + small constant for the Nothing sentinel - --- | Approximate byte cost of a cached NameRecord. Counts the user-controlled --- variable-length content plus a fixed per-entry overhead for the wrapper --- (TVar/PSQ node + ByteString headers + IORef). Tighter than a constant upper --- bound so cacheMaxBytes is a meaningful cap. -estimateBytes :: NameRecord -> Int -estimateBytes NameRecord {nrDisplayName, nrChannelLinks, nrContactLinks, nrAdminAddress, nrAdminEmail} = - perEntryOverhead - + utf8Len nrDisplayName - + 20 -- nrOwner - + sum (map nameLinkBytes nrChannelLinks) - + sum (map nameLinkBytes nrContactLinks) - + maybe 0 utf8Len nrAdminAddress - + maybe 0 utf8Len nrAdminEmail - where - perEntryOverhead = 256 -- PSQ node + key copy + ByteString headers - utf8Len = B.length . T.encodeUtf8 - nameLinkBytes :: NameLink -> Int - nameLinkBytes = utf8Len . unNameLink diff --git a/src/Simplex/Messaging/Server/Prometheus.hs b/src/Simplex/Messaging/Server/Prometheus.hs index f8a5f84bf..62d671224 100644 --- a/src/Simplex/Messaging/Server/Prometheus.hs +++ b/src/Simplex/Messaging/Server/Prometheus.hs @@ -461,7 +461,7 @@ prometheusMetrics sm rtm ts = \simplex_smp_" <> pfx <> "_services_sub_fewer_total " <> mshow (_srvSubFewerTotal ss) <> "\n# " <> pfx <> ".srvSubFewerTotal\n\ \\n" names = - let NameResolverStatsData {_rslvReqs, _rslvSucc, _rslvNotFound, _rslvEthErrs, _rslvCacheHits, _rslvCacheMiss, _rslvDisabled} = _rslvStats + let NameResolverStatsData {_rslvReqs, _rslvSucc, _rslvNotFound, _rslvEthErrs, _rslvDisabled} = _rslvStats in "# Names\n\ \# -----\n\ \\n\ @@ -481,14 +481,6 @@ prometheusMetrics sm rtm ts = \# TYPE simplex_smp_names_eth_errs counter\n\ \simplex_smp_names_eth_errs " <> mshow _rslvEthErrs <> "\n# rslvEthErrs\n\ \\n\ - \# HELP simplex_smp_names_cache_hits Resolution served from cache.\n\ - \# TYPE simplex_smp_names_cache_hits counter\n\ - \simplex_smp_names_cache_hits " <> mshow _rslvCacheHits <> "\n# rslvCacheHits\n\ - \\n\ - \# HELP simplex_smp_names_cache_miss Resolution required an eth_call.\n\ - \# TYPE simplex_smp_names_cache_miss counter\n\ - \simplex_smp_names_cache_miss " <> mshow _rslvCacheMiss <> "\n# rslvCacheMiss\n\ - \\n\ \# HELP simplex_smp_names_disabled RSLV requests rejected because the names role is disabled.\n\ \# TYPE simplex_smp_names_disabled counter\n\ \simplex_smp_names_disabled " <> mshow _rslvDisabled <> "\n# rslvDisabled\n\ diff --git a/src/Simplex/Messaging/Server/Stats.hs b/src/Simplex/Messaging/Server/Stats.hs index de9c23f19..84f5145b7 100644 --- a/src/Simplex/Messaging/Server/Stats.hs +++ b/src/Simplex/Messaging/Server/Stats.hs @@ -894,8 +894,6 @@ data NameResolverStats = NameResolverStats rslvSucc :: IORef Int, rslvNotFound :: IORef Int, rslvEthErrs :: IORef Int, - rslvCacheHits :: IORef Int, - rslvCacheMiss :: IORef Int, rslvDisabled :: IORef Int } @@ -905,18 +903,14 @@ newNameResolverStats = do rslvSucc <- newIORef 0 rslvNotFound <- newIORef 0 rslvEthErrs <- newIORef 0 - rslvCacheHits <- newIORef 0 - rslvCacheMiss <- newIORef 0 rslvDisabled <- newIORef 0 - pure NameResolverStats {rslvReqs, rslvSucc, rslvNotFound, rslvEthErrs, rslvCacheHits, rslvCacheMiss, rslvDisabled} + pure NameResolverStats {rslvReqs, rslvSucc, rslvNotFound, rslvEthErrs, rslvDisabled} data NameResolverStatsData = NameResolverStatsData { _rslvReqs :: Int, _rslvSucc :: Int, _rslvNotFound :: Int, _rslvEthErrs :: Int, - _rslvCacheHits :: Int, - _rslvCacheMiss :: Int, _rslvDisabled :: Int } deriving (Show) @@ -928,8 +922,6 @@ newNameResolverStatsData = _rslvSucc = 0, _rslvNotFound = 0, _rslvEthErrs = 0, - _rslvCacheHits = 0, - _rslvCacheMiss = 0, _rslvDisabled = 0 } @@ -939,10 +931,8 @@ getNameResolverStatsData s = do _rslvSucc <- readIORef $ rslvSucc s _rslvNotFound <- readIORef $ rslvNotFound s _rslvEthErrs <- readIORef $ rslvEthErrs s - _rslvCacheHits <- readIORef $ rslvCacheHits s - _rslvCacheMiss <- readIORef $ rslvCacheMiss s _rslvDisabled <- readIORef $ rslvDisabled s - pure NameResolverStatsData {_rslvReqs, _rslvSucc, _rslvNotFound, _rslvEthErrs, _rslvCacheHits, _rslvCacheMiss, _rslvDisabled} + pure NameResolverStatsData {_rslvReqs, _rslvSucc, _rslvNotFound, _rslvEthErrs, _rslvDisabled} getResetNameResolverStatsData :: NameResolverStats -> IO NameResolverStatsData getResetNameResolverStatsData s = do @@ -950,10 +940,8 @@ getResetNameResolverStatsData s = do _rslvSucc <- atomicSwapIORef (rslvSucc s) 0 _rslvNotFound <- atomicSwapIORef (rslvNotFound s) 0 _rslvEthErrs <- atomicSwapIORef (rslvEthErrs s) 0 - _rslvCacheHits <- atomicSwapIORef (rslvCacheHits s) 0 - _rslvCacheMiss <- atomicSwapIORef (rslvCacheMiss s) 0 _rslvDisabled <- atomicSwapIORef (rslvDisabled s) 0 - pure NameResolverStatsData {_rslvReqs, _rslvSucc, _rslvNotFound, _rslvEthErrs, _rslvCacheHits, _rslvCacheMiss, _rslvDisabled} + pure NameResolverStatsData {_rslvReqs, _rslvSucc, _rslvNotFound, _rslvEthErrs, _rslvDisabled} -- not thread safe; used on server start only setNameResolverStats :: NameResolverStats -> NameResolverStatsData -> IO () @@ -962,12 +950,10 @@ setNameResolverStats s d = do writeIORef (rslvSucc s) $! _rslvSucc d writeIORef (rslvNotFound s) $! _rslvNotFound d writeIORef (rslvEthErrs s) $! _rslvEthErrs d - writeIORef (rslvCacheHits s) $! _rslvCacheHits d - writeIORef (rslvCacheMiss s) $! _rslvCacheMiss d writeIORef (rslvDisabled s) $! _rslvDisabled d instance StrEncoding NameResolverStatsData where - strEncode NameResolverStatsData {_rslvReqs, _rslvSucc, _rslvNotFound, _rslvEthErrs, _rslvCacheHits, _rslvCacheMiss, _rslvDisabled} = + strEncode NameResolverStatsData {_rslvReqs, _rslvSucc, _rslvNotFound, _rslvEthErrs, _rslvDisabled} = "reqs=" <> strEncode _rslvReqs <> "\nsucc=" @@ -976,10 +962,6 @@ instance StrEncoding NameResolverStatsData where <> strEncode _rslvNotFound <> "\nethErrs=" <> strEncode _rslvEthErrs - <> "\ncacheHits=" - <> strEncode _rslvCacheHits - <> "\ncacheMiss=" - <> strEncode _rslvCacheMiss <> "\ndisabled=" <> strEncode _rslvDisabled strP = do @@ -987,10 +969,8 @@ instance StrEncoding NameResolverStatsData where _rslvSucc <- "succ=" *> strP <* A.endOfLine _rslvNotFound <- "notFound=" *> strP <* A.endOfLine _rslvEthErrs <- "ethErrs=" *> strP <* A.endOfLine - _rslvCacheHits <- "cacheHits=" *> strP <* A.endOfLine - _rslvCacheMiss <- "cacheMiss=" *> strP <* A.endOfLine _rslvDisabled <- "disabled=" *> strP - pure NameResolverStatsData {_rslvReqs, _rslvSucc, _rslvNotFound, _rslvEthErrs, _rslvCacheHits, _rslvCacheMiss, _rslvDisabled} + pure NameResolverStatsData {_rslvReqs, _rslvSucc, _rslvNotFound, _rslvEthErrs, _rslvDisabled} data ServiceStats = ServiceStats { srvAssocNew :: IORef Int, diff --git a/tests/SMPNamesTests.hs b/tests/SMPNamesTests.hs index f597c4ae0..102cf1f73 100644 --- a/tests/SMPNamesTests.hs +++ b/tests/SMPNamesTests.hs @@ -4,9 +4,6 @@ module SMPNamesTests (smpNamesTests) where -import Control.Concurrent (threadDelay) -import Control.Concurrent.Async (async, replicateConcurrently, wait) -import Control.Concurrent.STM (atomically, newEmptyTMVarIO, putTMVar, readTMVar) import qualified Crypto.Hash as Crypton import Data.ByteString.Char8 (ByteString) import qualified Data.ByteString.Char8 as B @@ -16,13 +13,13 @@ import Data.IORef (atomicModifyIORef', newIORef, readIORef) import qualified Data.Text as T import Simplex.Messaging.Encoding (smpEncode, smpP) import Simplex.Messaging.Parsers (parseAll) +import qualified Data.Aeson as J +import qualified Data.ByteString.Lazy as LB import Simplex.Messaging.Protocol ( LookupKey (..), NameRecord (..), mkNameLink, mkNameOwner, - nameRecBytes, - parseNameRec, unNameLink, unNameOwner, ) @@ -38,7 +35,7 @@ import Simplex.Messaging.Server.Names.Eth.SNRC namehash, snrcSelector, ) -import Simplex.Messaging.Server.Names.Resolver +import Simplex.Messaging.Server.Names ( NamesConfig (..), ResolveError (..), newNamesEnvWith, @@ -94,24 +91,23 @@ smpNamesTests = do describe "Keccak-256 and namehash" namehashSpec describe "ABI primitive bounds" abiBoundsSpec describe "decodeGetRecord (zero-owner sentinel)" zeroOwnerSpec - describe "Resolver cache + coalescing" resolverCacheSpec + describe "Resolver" resolverSpec nameRecordEncodingSpec :: Spec nameRecordEncodingSpec = do - it "round-trips nameRecBytes / parseNameRec" $ do - let bytes = nameRecBytes v20 sampleRecord - parseAll (parseNameRec v20) bytes `shouldBe` Right sampleRecord + it "round-trips JSON encode / decode" $ + J.eitherDecodeStrict (LB.toStrict (J.encode sampleRecord)) `shouldBe` Right sampleRecord it "rejects negative expiry" $ do - let badBytes = nameRecBytes v20 sampleRecord {nrExpiry = -1} - parseAll (parseNameRec v20) badBytes `shouldSatisfy` isLeft + let badBytes = LB.toStrict (J.encode sampleRecord {nrExpiry = -1}) + (J.eitherDecodeStrict badBytes :: Either String NameRecord) `shouldSatisfy` isLeft it "enforces combined channel+contact list cap of 8" $ do let mkLink i = either error id (mkNameLink ("simplex:/contact/" <> T.pack (show (i :: Int)))) nineLinks = map mkLink [0 .. 8] overflow = sampleRecord {nrChannelLinks = nineLinks, nrContactLinks = []} - bytes = nameRecBytes v20 overflow - parseAll (parseNameRec v20) bytes `shouldSatisfy` isLeft + bytes = LB.toStrict (J.encode overflow) + (J.eitherDecodeStrict bytes :: Either String NameRecord) `shouldSatisfy` isLeft it "encodes within the proxied transmission budget" $ do let huge = either error id (mkNameLink (T.replicate 1024 "x")) @@ -123,7 +119,7 @@ nameRecordEncodingSpec = do nrAdminAddress = Just (T.replicate 255 "a"), nrAdminEmail = Just (T.replicate 255 "e") } - B.length (nameRecBytes v20 wide) < 16224 `shouldBe` True + LB.length (J.encode wide) < 16224 `shouldBe` True lookupKeyAndCtorsSpec :: Spec lookupKeyAndCtorsSpec = do @@ -237,67 +233,31 @@ zeroOwnerSpec = do let tiny = B.replicate 31 '\NUL' decodeGetRecord tiny `shouldBe` Left AbiTruncated -resolverCacheSpec :: Spec -resolverCacheSpec = do +resolverSpec :: Spec +resolverSpec = do let mkEnv ethCall = do - hitsRef <- newIORef 0 - missRef <- newIORef 0 let cfg = NamesConfig { ethereumEndpoint = "http://stub", snrcAddress = either error id (mkNameOwner twentyOnes), rpcAuth = Nothing, - cacheSeconds = 300, - cacheMaxEntries = 100, - cacheMaxBytes = 1024 * 1024, rpcTimeoutMs = 1000, rpcMaxResponseBytes = 65536, rpcMaxConcurrency = 4 } - env <- newNamesEnvWith cfg ethCall Nothing hitsRef missRef - pure (env, hitsRef, missRef) + newNamesEnvWith cfg ethCall Nothing - it "maps stub zero-owner response to NotFound and counts as cache miss" $ do - (env, _, missRef) <- mkEnv $ \_ _ -> pure (Right (B.replicate (32 * 8) '\NUL')) + it "maps stub zero-owner response to NotFound" $ do + env <- mkEnv $ \_ _ -> pure (Right (B.replicate (32 * 8) '\NUL')) r <- resolveName env "alice" r `shouldBe` Left NotFound - misses <- readIORef missRef - misses `shouldBe` 1 - it "subsequent NotFound lookups hit the cache (no second RPC)" $ do + it "every lookup hits the endpoint (no cache)" $ do callCount <- newIORef (0 :: Int) - (env, hitsRef, missRef) <- mkEnv $ \_ _ -> do + env <- mkEnv $ \_ _ -> do atomicModifyIORef' callCount (\v -> (v + 1, ())) pure (Right (B.replicate (32 * 8) '\NUL')) - -- First lookup: miss, eth_call fires, NotFound cached. _ <- resolveName env "alice" - -- Second lookup: should hit cache, not call ethCall. - r2 <- resolveName env "alice" - r2 `shouldBe` Left NotFound - callCount' <- readIORef callCount - callCount' `shouldBe` 1 - missCount <- readIORef missRef - hitCount <- readIORef hitsRef - missCount `shouldBe` 1 - hitCount `shouldBe` 1 - - it "concurrent identical lookups coalesce — only the leader makes the RPC" $ do - -- Block the stub on a TMVar so the leader's eth_call doesn't return - -- before the 7 waiters race to attach to the inflight TMap. Without - -- coalescing, every caller would invoke ethCall and callCount would - -- be 8; with coalescing, only the leader fires. - gate <- newEmptyTMVarIO - callCount <- newIORef (0 :: Int) - (env, _, _) <- mkEnv $ \_ _ -> do - atomicModifyIORef' callCount (\v -> (v + 1, ())) - atomically (readTMVar gate) - pure (Right (B.replicate (32 * 8) '\NUL')) - -- Run the 8 callers in a background task so we can release the gate - -- only after they've all had a chance to register on the inflight map. - callers <- async $ replicateConcurrently 8 (resolveName env "alice") - threadDelay 50000 -- 50 ms — ample time for the 7 waiters to attach - atomically (putTMVar gate ()) - rs <- wait callers - all (== Left NotFound) rs `shouldBe` True + _ <- resolveName env "alice" n <- readIORef callCount - n `shouldBe` 1 + n `shouldBe` 2