mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-09-15 08:36:37 +00:00
shandEvgeny Poberezkin
c9ebf72e80
* tests: add SMP proxy relay reconnection tests Reproduces the proxy failing to reconnect to a destination relay when the sender disconnects mid-connection (empty session var left in smpClients). * fix: bracket session var creation to drop it on interrupt getSessVar inserts an empty session var that the connect path then fills with putTMVar. If the connecting thread is killed by an async exception before that fill (a proxy worker on client disconnect, an agent worker on cancel), the empty var was left in the map forever and every later request for that server blocked on it until timing out (permanent PCEResponseTimeout). Wrap get-or-create with withGetSessVar (bracketOnError) at the call sites, so the cleanup is established where the var is created and covers the whole connect: on interrupt before fill the still-empty var is dropped and the next request reconnects. This closes the window between getSessVar and the fill that a handler installed inside the connect function cannot cover. * test: cover session var leak on interrupted connect UtilTests: tryAllErrors rethrows ThreadKilled/StackOverflow (the mechanism that skips putTMVar). SMPProxyTests: agent client reconnection after a cancelled connect, plus a control proving the stalling relay alone does not cause the failure; refine the relay reconnection tests. * refactor --------- Co-authored-by: Evgeny Poberezkin <evgeny@poberezkin.com>
75 lines
3.0 KiB
Haskell
75 lines
3.0 KiB
Haskell
{-# LANGUAGE LambdaCase #-}
|
|
{-# LANGUAGE NamedFieldPuns #-}
|
|
{-# LANGUAGE ScopedTypeVariables #-}
|
|
|
|
module Simplex.Messaging.Session
|
|
( SessionVar (..),
|
|
getSessVar,
|
|
removeSessVar,
|
|
withGetSessVar,
|
|
withGetSessVar',
|
|
tryReadSessVar,
|
|
) where
|
|
|
|
import Control.Concurrent.STM
|
|
import Control.Monad.Except (ExceptT (..), runExceptT)
|
|
import Control.Monad.IO.Class (liftIO)
|
|
import Control.Monad.IO.Unlift (MonadUnliftIO)
|
|
import Data.Time (UTCTime)
|
|
import Simplex.Messaging.TMap (TMap)
|
|
import qualified Simplex.Messaging.TMap as TM
|
|
import Simplex.Messaging.Util (whenM, ($>>=))
|
|
import UnliftIO.Exception (bracketOnError)
|
|
|
|
data SessionVar a = SessionVar
|
|
{ sessionVar :: TMVar a,
|
|
sessionVarId :: Int,
|
|
sessionVarTs :: UTCTime
|
|
}
|
|
|
|
getSessVar :: forall k a. Ord k => TVar Int -> k -> TMap k (SessionVar a) -> UTCTime -> STM (Either (SessionVar a) (SessionVar a))
|
|
getSessVar sessSeq sessKey vs sessionVarTs = maybe (Left <$> newSessionVar) (pure . Right) =<< TM.lookup sessKey vs
|
|
where
|
|
newSessionVar :: STM (SessionVar a)
|
|
newSessionVar = do
|
|
sessionVar <- newEmptyTMVar
|
|
sessionVarId <- stateTVar sessSeq $ \next -> (next, next + 1)
|
|
let v = SessionVar {sessionVar, sessionVarId, sessionVarTs}
|
|
TM.insert sessKey v vs
|
|
pure v
|
|
|
|
removeSessVar :: Ord k => SessionVar a -> k -> TMap k (SessionVar a) -> STM ()
|
|
removeSessVar v sessKey vs =
|
|
TM.lookup sessKey vs >>= \case
|
|
Just v' | sessionVarId v == sessionVarId v' -> TM.delete sessKey vs
|
|
_ -> pure ()
|
|
|
|
-- | Get or create a session var and route to onNew (newly created) or onExisting. The new-var
|
|
-- branch is bracketed from the point of creation: if it is interrupted before filling the var
|
|
-- (e.g. an async exception during connect), the still-empty var is dropped from the map so the
|
|
-- next request creates a fresh session instead of blocking on a var that will never be filled.
|
|
-- A thrown ExceptT error is a normal result (the var keeps the error it was filled with) - only
|
|
-- an interrupting exception drops the empty var.
|
|
withGetSessVar ::
|
|
(Ord k, MonadUnliftIO m) =>
|
|
TVar Int -> k -> TMap k (SessionVar a) -> UTCTime ->
|
|
(SessionVar a -> ExceptT e m b) -> (SessionVar a -> ExceptT e m b) -> ExceptT e m b
|
|
withGetSessVar sessSeq sessKey vs ts onNew onExisting =
|
|
ExceptT $ withGetSessVar' sessSeq sessKey vs ts (runExceptT . onNew) (runExceptT . onExisting)
|
|
|
|
-- | withGetSessVar for actions in the underlying monad (without ExceptT).
|
|
withGetSessVar' ::
|
|
(Ord k, MonadUnliftIO m) =>
|
|
TVar Int -> k -> TMap k (SessionVar a) -> UTCTime ->
|
|
(SessionVar a -> m b) -> (SessionVar a -> m b) -> m b
|
|
withGetSessVar' sessSeq sessKey vs ts onNew onExisting =
|
|
bracketOnError
|
|
(liftIO $ atomically $ getSessVar sessSeq sessKey vs ts)
|
|
(either (liftIO . atomically . dropEmptySessVar) (\_ -> pure ()))
|
|
(either onNew onExisting)
|
|
where
|
|
dropEmptySessVar v = whenM (isEmptyTMVar $ sessionVar v) $ removeSessVar v sessKey vs
|
|
|
|
tryReadSessVar :: Ord k => k -> TMap k (SessionVar a) -> STM (Maybe a)
|
|
tryReadSessVar sessKey vs = TM.lookup sessKey vs $>>= (tryReadTMVar . sessionVar)
|