eventstore 0.14.0.0 → 0.14.0.1
raw patch · 5 files changed
+46/−23 lines, 5 filesdep ~aesonPVP ok
version bump matches the API change (PVP)
Dependency ranges changed: aeson
API changes (from Hackage documentation)
Files
- CHANGELOG.markdown +7/−0
- Database/EventStore/Internal/Execution/Production.hs +9/−9
- Database/EventStore/Internal/Subscription.hs +26/−10
- LICENSE +1/−1
- eventstore.cabal +3/−3
CHANGELOG.markdown view
@@ -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.
Database/EventStore/Internal/Execution/Production.hs view
@@ -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
Database/EventStore/Internal/Subscription.hs view
@@ -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
LICENSE view
@@ -1,4 +1,4 @@-Copyright (c) 2014, Yorick Laupa+Copyright (c) 2014-2017, Yorick Laupa All rights reserved.
eventstore.cabal view
@@ -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