diff --git a/CHANGELOG.markdown b/CHANGELOG.markdown --- a/CHANGELOG.markdown +++ b/CHANGELOG.markdown @@ -1,3 +1,10 @@ +0.14.0.1 +---------- +* Fix subscription deadlock on connection drop. +* Fix subscription catchup state tracking on connection drop. +* Fix internal connection dispose lock implementation. +* Bump `aeson` version. + 0.14.0.0 -------- * Fix deadlock issues in subscription code. diff --git a/Database/EventStore/Internal/Execution/Production.hs b/Database/EventStore/Internal/Execution/Production.hs --- a/Database/EventStore/Internal/Execution/Production.hs +++ b/Database/EventStore/Internal/Execution/Production.hs @@ -169,7 +169,7 @@ -- ^ Holds manager thread state. , _nextSubmit :: TVar (Msg -> IO ()) -- ^ Indicates the action to call in order to push new commands. - , _connRef :: IORef InternalConnection + , _connVar :: TVar InternalConnection -- ^ Connection to the server. , _disposed :: TMVar () -- ^ Indicates when the production execution model has been shutdown and @@ -350,7 +350,7 @@ -- | Spawns a new thread worker. spawn :: Env -> (ThreadId -> Worker) -> IO Worker spawn Env{..} mk = do - conn <- readIORef _connRef + conn <- readTVarIO _connVar tid <- mfix $ \tid -> let worker = mk tid action = @@ -517,7 +517,7 @@ -- If the connection is already closed, it will throw an exception. We just -- make sure it doesn't interfere with the cleaning process. - conn <- readIORef _connRef + conn <- readTVarIO _connVar _ <- try $ connClose conn :: (IO (Either ConnectionException ())) atomically $ do s <- readTVar _state @@ -552,11 +552,11 @@ pkg_queue <- newCycleQueue job_queue <- newCycleQueue conn <- newConnection setts disc - conn_ref <- newIORef conn + conn_var <- newTVarIO conn var <- newTVarIO $ emptyState setts gen nxt_sub <- newTVarIO (atomically . writeCycleQueue queue) disposed <- newEmptyTMVarIO - let env = Env setts queue pkg_queue job_queue var nxt_sub conn_ref disposed + let env = Env setts queue pkg_queue job_queue var nxt_sub conn_var disposed handler res = do closing env case res of @@ -570,18 +570,18 @@ , Handler $ \(ForceReconnectionException node) -> do new_conn <- newConnection setts disc connForceReconnect new_conn node - writeIORef conn_ref new_conn + atomically $ writeTVar conn_var new_conn _ <- forkFinally (bootstrap env) handler return () , Handler $ \(_ :: SomeException) -> do new_conn <- newConnection setts disc - writeIORef conn_ref new_conn + atomically $ writeTVar conn_var new_conn _ <- forkFinally (bootstrap env) handler return () ] _ -> atomically $ putTMVar disposed () _ <- forkFinally (bootstrap env) handler return $ Prod nxt_sub $ do - closed <- connIsClosed conn - unless closed retrySTM + curConn <- readTVar conn_var + unlessM (connIsClosed curConn) retrySTM readTMVar disposed diff --git a/Database/EventStore/Internal/Subscription.hs b/Database/EventStore/Internal/Subscription.hs --- a/Database/EventStore/Internal/Subscription.hs +++ b/Database/EventStore/Internal/Subscription.hs @@ -233,7 +233,7 @@ pushCmd = PushRegular streamId (catchupResLnkTos params) lcycle = SubLifeCycle - { onConfirm = atomically . putTMVar mvarRun + { onConfirm = confirmSub mvarRun , readState = readTMVar mvarState , writeState = \s -> () <$ swapTMVar mvarState s , onError = \e -> @@ -269,7 +269,7 @@ -> CatchupParams -> IO () tryRetryCatcupSubscription pushCmd env mvarState lcycle params = do - state <- atomically $ takeTMVar mvarState + state <- atomically $ readTMVar mvarState case state of -- In this case, we do our best to re-engage the @@ -295,7 +295,7 @@ newOp subPushConnect env (subEventHandler lcycle) pushCmd - _ -> atomically $ putTMVar mvarState state + _ -> return () -------------------------------------------------------------------------------- regularSub :: SubEnv -> Text -> Bool -> IO (Subscription Regular) @@ -304,7 +304,7 @@ varState <- newTVarIO $ SubOnline regularSubscription let lcycle = SubLifeCycle - { onConfirm = atomically . putTMVar mvarRun + { onConfirm = confirmSub mvarRun , readState = readTVar varState , writeState = writeTVar varState , onError = \r -> atomically $ do @@ -334,7 +334,7 @@ varState <- newTVarIO $ SubOnline persistentSubscription let lcycle = SubLifeCycle - { onConfirm = atomically . putTMVar mvarRun + { onConfirm = confirmSub mvarRun , readState = readTVar varState , writeState = writeTVar varState , onError = \r -> atomically $ do @@ -356,6 +356,16 @@ return $ Subscription stream lcycle env mvarRun (Persistent grp) -------------------------------------------------------------------------------- +-- | Makes sure to not cause deadlock because the subscription already been +-- confirmed but because of a connection drop, need to be recconfirmed again. +confirmSub :: TMVar Running -> Running -> IO () +confirmSub mvarRun r = atomically $ do + emptyVar <- isEmptyTMVar mvarRun + if emptyVar + then putTMVar mvarRun r + else () <$ swapTMVar mvarRun r + +-------------------------------------------------------------------------------- subNotHandledMsg :: SubEnv -> SubLifeCycle s -> NotHandledReason @@ -503,11 +513,17 @@ -> Checkpoint -> CatchupSMState -> CatchupSMState -insertReadEvents es chp s = - s { csmReadSeq = foldl' snoc (csmReadSeq s) es - , csmLiveSeq = dropWhileL (beforeChk chp) (csmLiveSeq s) - } +insertReadEvents es chp s = result + where + temp = s { csmReadSeq = foldl' snoc (csmReadSeq s) es + , csmLiveSeq = dropWhileL (beforeChk chp) (csmLiveSeq s) + } + result = + case chp of + CheckpointNumber n -> temp { csmLastNum = Just n } + CheckpointPosition p -> temp { csmLastPos = Just p } + -------------------------------------------------------------------------------- insertLiveEvent :: ResolvedEvent -> CatchupSMState -> CatchupSMState insertLiveEvent e s = s { csmLiveSeq = csmLiveSeq s `snoc` e } @@ -585,7 +601,7 @@ if isBatchReqEmpty nxtS then Just (e, SubStateMachine $ live s) else Just (e, SubStateMachine $ caughtUp nxtS) - caughtUp s BatchRead{} = SubStateMachine $ caughtUp s + caughtUp s input@BatchRead{} = catchingUp s input caughtUp _ CaughtUp = False caughtUp s LastEventNum = lastEventNumber s diff --git a/LICENSE b/LICENSE --- a/LICENSE +++ b/LICENSE @@ -1,4 +1,4 @@ -Copyright (c) 2014, Yorick Laupa +Copyright (c) 2014-2017, Yorick Laupa All rights reserved. diff --git a/eventstore.cabal b/eventstore.cabal --- a/eventstore.cabal +++ b/eventstore.cabal @@ -10,9 +10,9 @@ -- PVP summary: +-+------- breaking API changes -- | | +----- non-breaking API additions -- | | | +--- code changes with no API change -version: 0.14.0.0 +version: 0.14.0.1 -tested-with: GHC >= 7.8.3 && < 7.11 +tested-with: GHC >= 7.8.3 && < 8.0.1 -- A short (one-line) description of the package. synopsis: EventStore TCP Client @@ -106,7 +106,7 @@ -- Other library packages from which modules are imported. build-depends: base >=4.7 && <5 - , aeson >=0.8 && <1.1 + , aeson >=0.8 && <1.2 , cereal >=0.4 && <0.6 , containers ==0.5.* , protobuf >=0.2.1.1 && <0.3