diff --git a/CHANGELOG.markdown b/CHANGELOG.markdown --- a/CHANGELOG.markdown +++ b/CHANGELOG.markdown @@ -1,3 +1,12 @@ +1.4.0 +----- +* Better encoding of streaming interface ReadError. +* Expose more internal functions of the streaming interface. +* Internal: Reset heartbeat tracking on force reconnect request. +* Bug fix: Increment package number unless connection is init or closed (prevent false heartbeat timeout). +* Bug fix: Was also using `s_heartbeatInterval` value for `s_heartbeatTimeout` causing false heartbeat timeout exceptions. +* Bug fix: Stop trying to reconnect to the most recent node that failed (cluster connection mode). + 1.3.0 ----- * Discard `machines` dependency. diff --git a/Database/EventStore/Internal/ConnectionManager.hs b/Database/EventStore/Internal/ConnectionManager.hs --- a/Database/EventStore/Internal/ConnectionManager.hs +++ b/Database/EventStore/Internal/ConnectionManager.hs @@ -65,6 +65,18 @@ deriving Show -------------------------------------------------------------------------------- +atLeastEstablishingState :: Stage -> Bool +atLeastEstablishingState = \case + Init -> False + Connected{} -> True + Closed -> True + Connecting a s -> + case s of + Reconnecting -> False + EndpointDiscovery -> False + _ -> True + +-------------------------------------------------------------------------------- data Attempts = Attempts { attemptCount :: !Int , attemptLastStart :: !NominalDiffTime @@ -366,6 +378,10 @@ monitorIncrForceReconnect closeTcpConnection self (ForceReconnect ept) conn att <- freshAttempt _stopwatch + + -- We update the last pkg number to nullify the current heartbeat tracking. + atomicModifyIORef' _lastPkgNum $ \cur -> (cur + 1, ()) + atomicWriteIORef _stage (Connecting att EndpointDiscovery) $logInfo [i|#{conn}: going to reconnect to #{ept}.|] establish self ept @@ -415,9 +431,13 @@ Operation.check _opMgr atomicWriteIORef _lastCheck elapsed - manageHeartbeats self - _ -> return () + + -- FIXME - This `readIORef` call can be merged into the previous one. + -- Done in 2019's refactoring code. + stage <- readIORef _stage + when (atLeastEstablishingState stage) $ + manageHeartbeats self where onGoingConnection (Connecting att Reconnecting) = Just att onGoingConnection (Connecting att ConnectionEstablishing{}) = Just att @@ -461,7 +481,7 @@ setts <- getSettings let interval = s_heartbeatInterval setts - timeout = s_heartbeatInterval setts + timeout = s_heartbeatTimeout setts initTracker = tracker { _heartbeatStage = Interval , _startedSince = elapsed @@ -493,7 +513,14 @@ -------------------------------------------------------------------------------- onArrived :: Internal -> PackageArrived -> EventStore () -onArrived self@Internal{..} (PackageArrived conn pkg@Package{..}) = +onArrived self@Internal{..} (PackageArrived conn pkg@Package{..}) = do + -- FIXME - We can merge those two `readIORef` calls into one. + -- It's done on 2019's refactoring. + cur <- readIORef _stage + unless (closedOrInit cur) $ + incrPackageNumber self + -- /FIXME + readIORef _stage >>= \case (onAuthentication -> Just att) -> do when (packageCmd == notAuthenticatedCmd) $ @@ -506,7 +533,6 @@ (runningConnection -> True) -> do $logDebug [i|Package received: #{pkg}.|] - incrPackageNumber self handlePackage _ -> $logDebug [i|Package IGNORED: #{pkg}.|] @@ -541,6 +567,11 @@ case decision of Operation.Handled -> return () Operation.Reconnect node -> forceReconnect self node + + closedOrInit = \case + Init -> True + Closed -> True + _ -> False -------------------------------------------------------------------------------- isSameConnection :: Internal -> Connection -> EventStore Bool diff --git a/Database/EventStore/Internal/Discovery.hs b/Database/EventStore/Internal/Discovery.hs --- a/Database/EventStore/Internal/Discovery.hs +++ b/Database/EventStore/Internal/Discovery.hs @@ -361,8 +361,7 @@ case fend_m of Nothing -> oldGossip Just fend -> [ c | c <- oldGossip - , _externalTcpPort c == endPointPort fend - , _externalTcpIp c == endPointIp fend + , EndPoint (_externalTcpIp c) (_externalTcpPort c) /= fend ] -------------------------------------------------------------------------------- diff --git a/Database/EventStore/Streaming.hs b/Database/EventStore/Streaming.hs --- a/Database/EventStore/Streaming.hs +++ b/Database/EventStore/Streaming.hs @@ -1,4 +1,6 @@ -{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE GADTs #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE StandaloneDeriving #-} -------------------------------------------------------------------------------- -- | -- Module : Database.EventStore.Streaming @@ -11,9 +13,14 @@ -------------------------------------------------------------------------------- module Database.EventStore.Streaming ( ReadError(..) + , Fetch(..) + , ReadResultHandler(..) , readThroughForward , readThroughBackward , throwOnError + , defaultReadResultHandler + , onRegularStream + , readThrough ) where -------------------------------------------------------------------------------- @@ -34,33 +41,51 @@ import qualified Database.EventStore as ES -------------------------------------------------------------------------------- -data ReadError t - = StreamDeleted !ES.StreamName - | ReadError !(Maybe Text) - | AccessDenied !(ES.StreamId t) - deriving Show +data ReadError t where + StreamDeleted :: ES.StreamName -> ReadError ES.EventNumber + ReadError :: Maybe Text -> ReadError t + AccessDenied :: ES.StreamId t -> ReadError t + NoStream :: ReadError ES.EventNumber -------------------------------------------------------------------------------- +deriving instance Show (ReadError t) + +-------------------------------------------------------------------------------- instance (Show t, Typeable t) => Exception (ReadError t) -------------------------------------------------------------------------------- data Fetch t = FetchError !(ReadError t) | Fetch !(ES.Slice t) -------------------------------------------------------------------------------- -toFetch :: ES.ReadResult t (ES.Slice t) -> Fetch t -toFetch ES.ReadNoStream = Fetch ES.emptySlice -toFetch ES.ReadNotModified = Fetch ES.emptySlice -toFetch (ES.ReadStreamDeleted n) = FetchError (StreamDeleted n) -toFetch (ES.ReadError e) = FetchError (ReadError e) -toFetch (ES.ReadAccessDenied n) = FetchError (AccessDenied n) -toFetch (ES.ReadSuccess s) = Fetch s +newtype ReadResultHandler = + ReadResultHandler + { runReadResultHandler :: forall t. ES.StreamId t -> ES.BatchResult t -> Fetch t } -------------------------------------------------------------------------------- -handleBatchResult :: ES.StreamId t -> ES.BatchResult t -> Fetch t -handleBatchResult ES.StreamName{} = toFetch -handleBatchResult ES.All = Fetch +defaultReadResultHandler :: ReadResultHandler +defaultReadResultHandler = ReadResultHandler go + where + go :: ES.StreamId t -> ES.BatchResult t -> Fetch t + go ES.StreamName{} = toFetch + go ES.All = Fetch + toFetch ES.ReadNoStream = Fetch ES.emptySlice + toFetch ES.ReadNotModified = Fetch ES.emptySlice + toFetch (ES.ReadStreamDeleted n) = FetchError (StreamDeleted n) + toFetch (ES.ReadError e) = FetchError (ReadError e) + toFetch (ES.ReadAccessDenied n) = FetchError (AccessDenied n) + toFetch (ES.ReadSuccess s) = Fetch s + -------------------------------------------------------------------------------- +onRegularStream :: (ES.ReadResult ES.EventNumber (ES.Slice ES.EventNumber) -> Fetch ES.EventNumber) + -> ReadResultHandler +onRegularStream callback = ReadResultHandler go + where + go :: ES.StreamId t -> ES.BatchResult t -> Fetch t + go ES.StreamName{} = callback + go ES.All = Fetch + +-------------------------------------------------------------------------------- data State t = Need t | Fetched ![ES.ResolvedEvent] !(Maybe t) -------------------------------------------------------------------------------- @@ -93,7 +118,7 @@ -> Maybe Int32 -> Maybe ES.Credentials -> Stream (Of ES.ResolvedEvent) (ExceptT (ReadError t) IO) () -readThroughForward conn = readThrough conn ES.Forward +readThroughForward conn = readThrough conn defaultReadResultHandler ES.Forward -------------------------------------------------------------------------------- -- | Returns an iterator able to consume a stream entirely. When reading backward, @@ -105,7 +130,7 @@ -> Maybe Int32 -> Maybe ES.Credentials -> Stream (Of ES.ResolvedEvent) (ExceptT (ReadError t) IO) () -readThroughBackward conn = readThrough conn ES.Backward +readThroughBackward conn = readThrough conn defaultReadResultHandler ES.Backward -------------------------------------------------------------------------------- -- | Throws an exception in case 'ExceptT' is a 'Left'. @@ -120,7 +145,9 @@ Right a -> pure a -------------------------------------------------------------------------------- +-- | Returns an iterator able to consume a stream entirely. readThrough :: ES.Connection + -> ReadResultHandler -> ES.ReadDirection -> ES.StreamId t -> ES.ResolveLink @@ -128,35 +155,37 @@ -> Maybe Int32 -> Maybe ES.Credentials -> Stream (Of ES.ResolvedEvent) (ExceptT (ReadError t) IO) () -readThrough conn dir streamId lnk from sizMay cred = streaming iteratee from +readThrough conn handler dir streamId lnk from sizMay cred = streaming iteratee from where batchSize = fromMaybe 500 sizMay iteratee = case dir of - ES.Forward -> readForward conn streamId batchSize lnk cred - ES.Backward -> readBackward conn streamId batchSize lnk cred + ES.Forward -> readForward conn handler streamId batchSize lnk cred + ES.Backward -> readBackward conn handler streamId batchSize lnk cred -------------------------------------------------------------------------------- readForward :: ES.Connection + -> ReadResultHandler -> ES.StreamId t -> Int32 -> ES.ResolveLink -> Maybe ES.Credentials -> t -> IO (Fetch t) -readForward conn streamId siz lnk creds start = - fmap (handleBatchResult streamId) . wait =<< +readForward conn handler streamId siz lnk creds start = + fmap (runReadResultHandler handler streamId) . wait =<< ES.readEventsForward conn streamId start siz lnk creds -------------------------------------------------------------------------------- readBackward :: ES.Connection + -> ReadResultHandler -> ES.StreamId t -> Int32 -> ES.ResolveLink -> Maybe ES.Credentials -> t -> IO (Fetch t) -readBackward conn streamId siz lnk creds start = - fmap (handleBatchResult streamId) . wait =<< +readBackward conn handler streamId siz lnk creds start = + fmap (runReadResultHandler handler streamId) . wait =<< ES.readEventsBackward conn streamId start siz lnk creds diff --git a/eventstore.cabal b/eventstore.cabal --- a/eventstore.cabal +++ b/eventstore.cabal @@ -4,14 +4,14 @@ -- -- see: https://github.com/sol/hpack -- --- hash: 663b08f5bc970725b2a57c47aa571a768636e650f33cdb9b1d1a1b3d0d351dd4 +-- hash: 1183167359a5b717d35d6bbd4ca75dcce93e5223d77d575319cd5fe83124a473 name: eventstore -version: 1.3.0 +version: 1.3.1 synopsis: EventStore TCP Client description: EventStore TCP Client category: Database -homepage: https://gitlab.com/YoEight/eventstore-hs +homepage: https://github.com/YoEight/eventstore author: Yorick Laupa maintainer: yo.eight@gmail.com copyright: Yorick Laupa @@ -25,7 +25,7 @@ source-repository head type: git - location: https://gitlab.com/YoEight/eventstore-hs.git + location: https://github.com/YoEight/eventstore.git library exposed-modules: