mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-10-23 00:02:44 +00:00
smp-server: fuse parallel dispatchers
This commit is contained in:
1 parent
3999facde3
commit
99b1330dc5
2 files changed
+33
-31
No files matched your search
@@ -65,7 +65,7 @@ import Data.Constraint (Dict (..))
|
||||
import Data.Dynamic (toDyn)
|
||||
import Data.Either (fromRight, partitionEithers)
|
||||
import Data.Foldable (foldrM)
|
||||
import Data.Functor (($>))
|
||||
import Data.Functor (($>), (<&>))
|
||||
import Data.IORef
|
||||
import Data.Int (Int64)
|
||||
import qualified Data.IntMap.Strict as IM
|
||||
@@ -1499,13 +1499,13 @@ client
|
||||
Cmd SResolver (RSLV (LookupKey key)) -> do
|
||||
st <- asks (rslvStats . serverStats)
|
||||
incStat (rslvReqs st)
|
||||
asks namesEnv >>= \case
|
||||
Nothing -> incStat (rslvDisabled st) $> response (corrId, NoEntity, ERR AUTH)
|
||||
Just nenv ->
|
||||
liftIO (resolveName nenv key) >>= \case
|
||||
Right rec -> incStat (rslvSucc st) $> response (corrId, NoEntity, NAME rec)
|
||||
Left NotFound -> incStat (rslvNotFound st) $> response (corrId, NoEntity, ERR AUTH)
|
||||
Left _ -> incStat (rslvEthErrs st) $> response (corrId, NoEntity, ERR AUTH)
|
||||
(selector, msg) <- asks namesEnv >>= \case
|
||||
Nothing -> pure (rslvDisabled, ERR AUTH)
|
||||
Just nenv -> liftIO (resolveName nenv key) <&> \case
|
||||
Right rec -> (rslvSucc, NAME rec)
|
||||
Left NotFound -> (rslvNotFound, ERR AUTH)
|
||||
Left _ -> (rslvEthErrs, ERR AUTH)
|
||||
incStat (selector st) $> response (corrId, NoEntity, msg)
|
||||
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
|
||||
|
||||
@@ -195,7 +195,7 @@ coalesce env@NamesEnv {inflight} key now = do
|
||||
E.throwIO e
|
||||
| otherwise -> do
|
||||
logError $ "[NAMES] resolver fetch raised " <> T.pack (E.displayException e)
|
||||
pure (Left (mapSyncEthExn e))
|
||||
pure (Left EthHttpErr)
|
||||
atomically $ do
|
||||
putTMVar mv r
|
||||
modifyTVar' inflight (PSQ.delete key)
|
||||
@@ -205,9 +205,6 @@ coalesce env@NamesEnv {inflight} key now = do
|
||||
Left _ -> pure () -- transient errors (HTTP, decode, timeout) are not cached
|
||||
pure r
|
||||
|
||||
mapSyncEthExn :: E.SomeException -> ResolveError
|
||||
mapSyncEthExn _ = EthHttpErr
|
||||
|
||||
fetchOnceTimed :: NamesEnv -> ByteString -> IO (Either ResolveError NameRecord)
|
||||
fetchOnceTimed env key =
|
||||
timeout (rpcTimeoutMs (config env) * 1000) (fetchOnce env key) >>= \case
|
||||
@@ -215,28 +212,33 @@ fetchOnceTimed env key =
|
||||
Nothing -> pure (Left TimedOut)
|
||||
|
||||
fetchOnce :: NamesEnv -> ByteString -> IO (Either ResolveError NameRecord)
|
||||
fetchOnce NamesEnv {ethCall, config} key = do
|
||||
let node = namehash key
|
||||
callData = encodeGetRecord node
|
||||
to = unNameOwner (snrcAddress config)
|
||||
ethCall to callData >>= \case
|
||||
Left (HttpFailure _) -> pure (Left EthHttpErr)
|
||||
Left (HttpStatusErr _) -> pure (Left EthHttpErr)
|
||||
Left BodyTooLarge -> pure (Left EthDecodeErr)
|
||||
Left (InvalidJson _) -> pure (Left EthDecodeErr)
|
||||
Left (JsonRpcErr c m) -> pure (Left EthRpcErr {rpcCode = c, rpcMessage = m})
|
||||
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) -> do
|
||||
nowSec <- floor <$> getPOSIXTime
|
||||
-- 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.
|
||||
if nrExpiry rec /= 0 && nrExpiry rec < nowSec
|
||||
then pure (Left NotFound)
|
||||
else pure (Right rec)
|
||||
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
|
||||
|
||||
Reference in new issue
Block a user