eventstore 1.2.1 → 1.2.2

raw patch · 7 files changed

+196/−72 lines, 7 filesdep +file-embeddep +vectorPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: file-embed, vector

API changes (from Hackage documentation)

Files

• CHANGELOG.markdown +9/−0
• Database/EventStore/Internal/ConnectionManager.hs +8/−1
• Database/EventStore/Internal/Discovery.hs +28/−22
• Database/EventStore/Internal/Manager/Operation/Registry.hs +7/−0
• Database/EventStore/Internal/Types.hs +77/−47
• eventstore.cabal +5/−2
• tests/Test/Integration/Tests.hs +62/−0

@@ -1,3 +1,12 @@+1.2.2+-----+* Fix ACL JSON (de)serialization.+* Fix rare situation where the user send a command before the connection with the+  server is available, causing that operation to be sent only after `s_operationTimeout` milliseconds.+* Fix "thread blocked indefinitely in an STM transaction" error when failed to create a+  connection with the server and exceeded the maximun attempt count.+* Fix: Cluster connection, in the discovery process using gossip seed, we try for the next seed if the current one is unreachable.+ 1.2.1 ----- * Bump `http-client` version.

@@ -217,7 +217,7 @@           atomicWriteIORef _stage (Connecting att EndpointDiscovery)           old <- readIORef _last           _   <- fork $-              tryAny (liftIO $ runDiscovery _disc old) >>= \case+              tryAny (runDiscovery _disc old) >>= \case                 Left e -> do                   $logError                     [i| Failed to resolve TCP endpoint to which to connect #{e}.|]@@ -300,6 +300,13 @@       $logDebug [i|TCP connection identified: #{conn}.|]       atomicWriteIORef _stage (Connected conn)       initHeartbeatTracker self++      -- HACK: It can happen the user submitted operations before the connection was+      -- available. Those operations are only check on every 's_operationTimeout'+      -- ms. This could lead the first operation to take time before gettings.+      -- FIXME: We might consider doing that hack only if it's the first time+      -- we connect with the server.+      Operation.check _opMgr     _ -> pure ()  --------------------------------------------------------------------------------

@@ -38,6 +38,7 @@ import Data.Maybe  --------------------------------------------------------------------------------+import Control.Exception.Safe (SomeException, tryAny) import Data.Aeson import Data.Aeson.Types import Data.Array.IO@@ -49,7 +50,9 @@ import System.Random  --------------------------------------------------------------------------------+import Database.EventStore.Internal.Control import Database.EventStore.Internal.EndPoint+import Database.EventStore.Internal.Logger import Database.EventStore.Internal.Prelude  --------------------------------------------------------------------------------@@ -107,7 +110,7 @@ -------------------------------------------------------------------------------- -- | Procedure used to discover an network 'EndPoint'. newtype Discovery =-    Discovery { runDiscovery :: Maybe EndPoint -> IO (Maybe EndPoint) }+    Discovery { runDiscovery :: Maybe EndPoint -> EventStore (Maybe EndPoint) }  -------------------------------------------------------------------------------- staticEndPointDiscovery :: String -> Int -> Discovery@@ -127,8 +130,8 @@                                 DnsHostName h   -> RCHostName h                                 DnsHostPort h p -> RCHostPort h (fromIntegral p)                     in defaultResolvConf { resolvInfo = rc }-    dnsSeed <- makeResolvSeed conf-    res     <- withResolver dnsSeed $ \resv -> lookupA resv domain+    dnsSeed <- liftIO $ makeResolvSeed conf+    res     <- liftIO $ withResolver dnsSeed $ \resv -> lookupA resv domain     case res of         Left e    -> throwIO $ DNSDiscoveryError e         Right ips -> do@@ -292,17 +295,16 @@                  -> IORef (Maybe [MemberInfo])                  -> Maybe EndPoint                  -> ClusterSettings-                 -> IO (Maybe EndPoint)+                 -> EventStore (Maybe EndPoint) discoverEndPoint mgr ref fend settings = do     old_m <- readIORef ref     writeIORef ref Nothing     candidates <- case old_m of         Nothing  -> gossipCandidatesFromDns settings-        Just old -> gossipCandidatesFromOldGossip fend old+        Just old -> liftIO $ gossipCandidatesFromOldGossip fend old     forArrayFirst candidates $ \i -> do-        c   <- readArray candidates i+        c   <- liftIO $ readArray candidates i         res <- tryGetGossipFrom settings mgr c-        print (i, res)         let fin_end = do                 info <- res                 best <- tryDetermineBestNode $ members info@@ -317,13 +319,17 @@ tryGetGossipFrom :: ClusterSettings                  -> Manager                  -> GossipSeed-                 -> IO (Maybe ClusterInfo)+                 -> EventStore (Maybe ClusterInfo) tryGetGossipFrom ClusterSettings{..} mgr seed = do-    init_req <- httpRequest (gossipEndpoint seed) "/gossip?format=json"+    init_req <- liftIO $ httpRequest (gossipEndpoint seed) "/gossip?format=json"     let timeout = truncate (totalMillis clusterGossipTimeout * 1000)         req     = init_req { responseTimeout = responseTimeoutMicro timeout }-    resp <- httpLbs req mgr-    return $ decode $ responseBody resp+    eithResp <- tryAny $ liftIO $ httpLbs req mgr+    case eithResp of+        Right resp -> return $ decode $ responseBody resp+        Left err   -> do+            $logInfo [i|Failed to get cluster info from [#{seed}], error: #{err}.|]+            pure Nothing  -------------------------------------------------------------------------------- tryDetermineBestNode :: [MemberInfo] -> Maybe EndPoint@@ -391,10 +397,10 @@         seed = GossipSeed end ""  ---------------------------------------------------------------------------------gossipCandidatesFromDns :: ClusterSettings -> IO (IOArray Int GossipSeed)+gossipCandidatesFromDns :: ClusterSettings -> EventStore (IOArray Int GossipSeed) gossipCandidatesFromDns settings@ClusterSettings{..} = do     arr <- endpoints-    shuffleAll arr+    liftIO $ shuffleAll arr     return arr   where     endpoints =@@ -402,10 +408,10 @@             Nothing -> resolveDns settings             Just ss -> let ls  = toList ss                            len = length ls-                  in newListArray (0, len - 1) ls+                  in liftIO $ newListArray (0, len - 1) ls  ---------------------------------------------------------------------------------resolveDns :: ClusterSettings -> IO (IOArray Int GossipSeed)+resolveDns :: ClusterSettings -> EventStore (IOArray Int GossipSeed) resolveDns ClusterSettings{..} = do     let timeoutMicros = totalMillis clusterGossipTimeout * 1000         conf =@@ -418,11 +424,11 @@                                 DnsHostName h   -> RCHostName h                                 DnsHostPort h p -> RCHostPort h (fromIntegral p)                     in defaultResolvConf { resolvInfo = rc  }-    dnsSeed <- makeResolvSeed conf+    dnsSeed <- liftIO $ makeResolvSeed conf                { resolvTimeout = truncate timeoutMicros                , resolvRetry   = clusterMaxDiscoverAttempts                }-    withResolver dnsSeed $ \resv -> do+    liftIO $ withResolver dnsSeed $ \resv -> do         result <- lookupA resv clusterDns         case result of             Left e    -> throwIO $ DNSDiscoveryError e@@ -463,17 +469,17 @@  -------------------------------------------------------------------------------- forArrayFirst :: IOArray Int a-              -> (Int -> IO (Maybe b))-              -> IO (Maybe b)+              -> (Int -> EventStore (Maybe b))+              -> EventStore (Maybe b) forArrayFirst arr k = do-    (low, hig) <- getBounds arr+    (low, hig) <- liftIO $ getBounds arr     forRangeFirst low hig k  -------------------------------------------------------------------------------- forRangeFirst :: Int               -> Int-              -> (Int -> IO (Maybe b))-              -> IO (Maybe b)+              -> (Int -> EventStore (Maybe b))+              -> EventStore (Maybe b) forRangeFirst from to k = do     if from <= to then loop (to + 1) from else return Nothing   where

@@ -179,6 +179,11 @@ rejectPending Pending{..} = rejectSession _pendingSession  --------------------------------------------------------------------------------+rejectAwaiting :: Exception e => Awaiting -> e -> EventStore ()+rejectAwaiting (Awaiting s) = rejectSession s+rejectAwaiting (AwaitingRequest s _) = rejectSession s++-------------------------------------------------------------------------------- applyResponse :: Registry -> Pending -> Package -> EventStore () applyResponse reg Pending{..} = resumeSession reg _pendingSession @@ -300,8 +305,10 @@ abortPendingRequests :: Registry -> EventStore () abortPendingRequests Registry{..} = do   m <- atomicModifyIORef' _regPendings $ \pendings -> (mempty, pendings)+  ws <- atomicModifyIORef' _regAwaitings $ \aw -> (mempty, aw)    for_ m $ \p -> rejectPending p Aborted+  for_ ws $ \a -> rejectAwaiting a Aborted  -------------------------------------------------------------------------------- data Decision

@@ -36,6 +36,7 @@ import           Data.Time (NominalDiffTime) import           Data.Time.Clock.POSIX import           Data.UUID (UUID, fromByteString, toByteString)+import qualified Data.Vector as Vector  -------------------------------------------------------------------------------- import Database.EventStore.Internal.Command@@ -666,17 +667,17 @@ -- | Represents an access control list for a stream. data StreamACL     = StreamACL-      { streamACLReadRoles :: ![Text]+      { streamACLReadRoles :: !(Maybe [Text])         -- ^ Roles and users permitted to read the stream.-      , streamACLWriteRoles :: ![Text]+      , streamACLWriteRoles :: !(Maybe [Text])         -- ^ Roles and users permitted to write to the stream.-      , streamACLDeleteRoles :: ![Text]+      , streamACLDeleteRoles :: !(Maybe [Text])         -- ^ Roles and users permitted to delete to the stream.-      , streamACLMetaReadRoles :: ![Text]+      , streamACLMetaReadRoles :: !(Maybe [Text])         -- ^ Roles and users permitted to read stream metadata.-      , streamACLMetaWriteRoles :: ![Text]+      , streamACLMetaWriteRoles :: !(Maybe [Text])         -- ^ Roles and users permitted to write stream metadata.-      } deriving Show+      } deriving (Show, Eq)  ------------------------------------------------------------------------------- instance A.FromJSON StreamACL where@@ -690,11 +691,11 @@ -- | 'StreamACL' with no role or users whatsoever. emptyStreamACL :: StreamACL emptyStreamACL = StreamACL-                 { streamACLReadRoles      = []-                 , streamACLWriteRoles     = []-                 , streamACLDeleteRoles    = []-                 , streamACLMetaReadRoles  = []-                 , streamACLMetaWriteRoles = []+                 { streamACLReadRoles      = Nothing+                 , streamACLWriteRoles     = Nothing+                 , streamACLDeleteRoles    = Nothing+                 , streamACLMetaReadRoles  = Nothing+                 , streamACLMetaWriteRoles = Nothing                  }  --------------------------------------------------------------------------------@@ -711,12 +712,12 @@         --   is used to implement soft-deletion of streams.       , streamMetadataCacheControl :: !(Maybe TimeSpan)         -- ^ The amount of time for which the stream head is cachable.-      , streamMetadataACL :: !StreamACL+      , streamMetadataACL :: !(Maybe StreamACL)         -- ^ The access control list for the stream.       , streamMetadataCustom :: !Object         -- ^ An enumerable of key-value pairs of keys to JSON text for         --   user-provider metadata.-      } deriving Show+      } deriving (Show, Eq)  -------------------------------------------------------------------------------- -- | Gets a custom property value from metadata.@@ -751,7 +752,7 @@                       , streamMetadataMaxAge         = Nothing                       , streamMetadataTruncateBefore = Nothing                       , streamMetadataCacheControl   = Nothing-                      , streamMetadataACL            = emptyStreamACL+                      , streamMetadataACL            = Nothing                       , streamMetadataCustom         = mempty                       } @@ -763,26 +764,45 @@     go (k,v) = k .= v  --------------------------------------------------------------------------------+-- | Gets rid of null-ed properties. If a value is an array and that array only+--   has one element, that function simplifies that array of JSON to a single+--   JSON value.+cleanPairs :: [Pair] -> [Pair]+cleanPairs xs = xs >>= go+  where+    go (_, A.Null) = []+    go (name, obj) = [(name, deeper obj)]++    deeper cur@(A.Array xs)+      | Vector.length xs == 1 = Vector.head xs+      | otherwise             = cur+    deeper cur = cur++-------------------------------------------------------------------------------- -- | Serialized a 'StreamACL' to 'Value' for serialization purpose. streamACLJSON :: StreamACL -> A.Value streamACLJSON StreamACL{..} =-    A.object [ p_readRoles      .= streamACLReadRoles-             , p_writeRoles     .= streamACLWriteRoles-             , p_deleteRoles    .= streamACLDeleteRoles-             , p_metaReadRoles  .= streamACLMetaReadRoles-             , p_metaWriteRoles .= streamACLMetaWriteRoles-             ]+    A.object $+        cleanPairs+        [ p_readRoles      .= streamACLReadRoles+        , p_writeRoles     .= streamACLWriteRoles+        , p_deleteRoles    .= streamACLDeleteRoles+        , p_metaReadRoles  .= streamACLMetaReadRoles+        , p_metaWriteRoles .= streamACLMetaWriteRoles+        ]  -------------------------------------------------------------------------------- -- | Serialized a 'StreamMetadata' to 'Value' for serialization purpose. streamMetadataJSON :: StreamMetadata -> A.Value streamMetadataJSON StreamMetadata{..} =-    A.object $ [ p_maxAge         .= fmap toInt64 streamMetadataMaxAge-               , p_maxCount       .= streamMetadataMaxCount-               , p_truncateBefore .= streamMetadataTruncateBefore-               , p_cacheControl   .= fmap toInt64 streamMetadataCacheControl-               , p_acl            .= streamACLJSON streamMetadataACL-               ] <> custPairs+    A.object $+        cleanPairs+        [ p_maxAge         .= fmap toInt64 streamMetadataMaxAge+        , p_maxCount       .= streamMetadataMaxCount+        , p_truncateBefore .= streamMetadataTruncateBefore+        , p_cacheControl   .= fmap toInt64 streamMetadataCacheControl+        , p_acl            .= fmap streamACLJSON streamMetadataACL+        ] <> custPairs   where     custPairs = customMetaToPairs streamMetadataCustom @@ -873,25 +893,35 @@ -- | Parses 'StreamACL'. parseStreamACL :: A.Value -> Parser StreamACL parseStreamACL (A.Object m) =-    StreamACL              <$>-    m A..: p_readRoles     <*>-    m A..: p_writeRoles    <*>-    m A..: p_deleteRoles   <*>-    m A..: p_metaReadRoles <*>-    m A..: p_metaWriteRoles+    StreamACL+        <$> parseSingleOrMultiple m p_readRoles+        <*> parseSingleOrMultiple m p_writeRoles+        <*> parseSingleOrMultiple m p_deleteRoles+        <*> parseSingleOrMultiple m p_metaReadRoles+        <*> parseSingleOrMultiple m p_metaWriteRoles parseStreamACL _ = mzero  --------------------------------------------------------------------------------+parseSingleOrMultiple :: A.Object -> Text -> Parser (Maybe [Text])+parseSingleOrMultiple obj name = multiple <|> single+  where+    single = do+        mV <- obj A..: name <|> pure Nothing+        pure $ fmap (\v -> [v]) mV++    multiple = obj A..: name++-------------------------------------------------------------------------------- -- | Parses 'StreamMetadata'. parseStreamMetadata :: A.Value -> Parser StreamMetadata parseStreamMetadata (A.Object m) =-    StreamMetadata                    <$>-    m A..: p_maxCount                 <*>-    parseTimeSpan p_maxAge            <*>-    m A..: p_truncateBefore           <*>-    parseTimeSpan p_cacheControl      <*>-    (m A..: p_acl >>= parseStreamACL) <*>-    pure (keepUserProperties m)+    StreamMetadata+        <$> (m A..: p_maxCount <|> pure Nothing)+        <*> (parseTimeSpan p_maxAge <|> pure Nothing)+        <*> (m A..: p_truncateBefore <|> pure Nothing)+        <*> (parseTimeSpan p_cacheControl <|> pure Nothing)+        <*> ((m A..: p_acl >>= traverse parseStreamACL) <|> pure Nothing)+        <*> pure (keepUserProperties m)   where     parseTimeSpan ::  Text -> Parser (Maybe TimeSpan)     parseTimeSpan prop = do@@ -917,7 +947,7 @@ -------------------------------------------------------------------------------- -- | Sets role names with read permission for the stream. setReadRoles :: [Text] -> StreamACLBuilder-setReadRoles xs = Endo $ \s -> s { streamACLReadRoles = xs }+setReadRoles xs = Endo $ \s -> s { streamACLReadRoles = Just xs }  -------------------------------------------------------------------------------- -- | Sets a single role name with read permission for the stream.@@ -927,7 +957,7 @@ -------------------------------------------------------------------------------- -- | Sets role names with write permission for the stream. setWriteRoles :: [Text] -> StreamACLBuilder-setWriteRoles xs = Endo $ \s -> s { streamACLWriteRoles = xs }+setWriteRoles xs = Endo $ \s -> s { streamACLWriteRoles = Just xs }  -------------------------------------------------------------------------------- -- | Sets a single role name with write permission for the stream.@@ -937,7 +967,7 @@ -------------------------------------------------------------------------------- -- | Sets role names with delete permission for the stream. setDeleteRoles :: [Text] -> StreamACLBuilder-setDeleteRoles xs = Endo $ \s -> s { streamACLDeleteRoles = xs }+setDeleteRoles xs = Endo $ \s -> s { streamACLDeleteRoles = Just xs }  -------------------------------------------------------------------------------- -- | Sets a single role name with delete permission for the stream.@@ -947,7 +977,7 @@ -------------------------------------------------------------------------------- -- | Sets role names with metadata read permission for the stream. setMetaReadRoles :: [Text] -> StreamACLBuilder-setMetaReadRoles xs = Endo $ \s -> s { streamACLMetaReadRoles = xs }+setMetaReadRoles xs = Endo $ \s -> s { streamACLMetaReadRoles = Just xs }  -------------------------------------------------------------------------------- -- | Sets a single role name with metadata read permission for the stream.@@ -957,7 +987,7 @@ -------------------------------------------------------------------------------- -- | Sets role names with metadata write permission for the stream. setMetaWriteRoles :: [Text] -> StreamACLBuilder-setMetaWriteRoles xs = Endo $ \s -> s { streamACLMetaWriteRoles = xs }+setMetaWriteRoles xs = Endo $ \s -> s { streamACLMetaWriteRoles = Just xs }  -------------------------------------------------------------------------------- -- | Sets a single role name with metadata write permission for the stream.@@ -1002,14 +1032,14 @@ -- | Overwrites any previous 'StreamACL' by the given one in a --   'StreamMetadataBuilder'. setACL :: StreamACL -> StreamMetadataBuilder-setACL a = Endo $ \s -> s { streamMetadataACL = a }+setACL a = Endo $ \s -> s { streamMetadataACL = Just a }  -------------------------------------------------------------------------------- -- | Updates a 'StreamMetadata''s 'StreamACL' given a 'StreamACLBuilder'. modifyACL :: StreamACLBuilder -> StreamMetadataBuilder modifyACL b = Endo $ \s ->-    let old = streamMetadataACL s-    in s { streamMetadataACL = modifyStreamACL b old }+    let old = fromMaybe emptyStreamACL $ streamMetadataACL s+    in s { streamMetadataACL = Just $ modifyStreamACL b old }  -------------------------------------------------------------------------------- -- | Sets a custom metadata property.

@@ -4,10 +4,10 @@ -- -- see: https://github.com/sol/hpack ----- hash: 89315b698f4d189670b511052a2a9867110322bf62663f0462d64a2bc314e65b+-- hash: 37d679d4881e9e80b3b25d6599d421f90635c1cd0f73703bd4c1b18b32f318b5  name:           eventstore-version:        1.2.1+version:        1.2.2 synopsis:       EventStore TCP Client description:    EventStore TCP Client <https://eventstore.org> category:       Database@@ -125,6 +125,7 @@     , transformers-base     , unordered-containers     , uuid ==1.3.*+    , vector   default-language: Haskell2010  test-suite eventstore-tests@@ -155,6 +156,7 @@     , eventstore     , exceptions     , fast-logger+    , file-embed     , hashable     , lifted-async     , lifted-base@@ -175,4 +177,5 @@     , transformers-base     , unordered-containers     , uuid+    , vector   default-language: Haskell2010

@@ -1,6 +1,7 @@ {-# LANGUAGE GADTs               #-} {-# LANGUAGE OverloadedStrings   #-} {-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TemplateHaskell     #-} -------------------------------------------------------------------------------- -- | -- Module : Test.Integration.Tests@@ -19,6 +20,7 @@ import Control.Concurrent.Async (wait) import Data.Aeson import Data.DotNet.TimeSpan+import Data.FileEmbed (embedFile) import Data.Maybe (fromMaybe) import Data.UUID hiding (null) import Data.UUID.V4@@ -111,6 +113,11 @@     it "creates a volatile subscription" subscribeTest     it "creates a catchup subscription" subscribeFromTest     it "proves catchup subscriptions don't deadlock on non existant streams" subscribeFromNoStreamTest+    it "parses empty stream acl" emptyAclParsing+    it "parses single ACL property" aclSingleReadParsing+    it "parses multiple ACL property" aclMultipleReadParsing+    it "parses empty stream meta" emptyStreamMetaParsing+    it "parses stream meta property" metaPropParsing     it "sets stream metadata" setStreamMetadataTest     it "gets stream metadata" getStreamMetadataTest     it "creates a persistent subscription" createPersistentTest@@ -376,6 +383,61 @@                 Just i -> assertEqual "Should have equal value" (1 :: Int) i                 _      -> fail "Can't find foo property"         _ -> fail $ "Stream " <> show stream <> " doesn't exist"++--------------------------------------------------------------------------------+emptyAclBytes :: ByteString+emptyAclBytes = $(embedFile "tests/fixtures/empty_acl.json")++--------------------------------------------------------------------------------+emptyMetaBytes :: ByteString+emptyMetaBytes = $(embedFile "tests/fixtures/empty_meta.json")++--------------------------------------------------------------------------------+aclSingleReadBytes :: ByteString+aclSingleReadBytes = $(embedFile "tests/fixtures/acl_single_read.json")++--------------------------------------------------------------------------------+aclMultipleReadBytes :: ByteString+aclMultipleReadBytes = $(embedFile "tests/fixtures/acl_multiple_read.json")++--------------------------------------------------------------------------------+metaPropBytes :: ByteString+metaPropBytes = $(embedFile "tests/fixtures/meta_prop.json")++--------------------------------------------------------------------------------+emptyAclParsing :: a -> IO ()+emptyAclParsing _ =+    assertEqual "should parse empty ACL"+        (Just emptyStreamACL)+        (decode $ fromStrict emptyAclBytes)++--------------------------------------------------------------------------------+aclSingleReadParsing :: a -> IO ()+aclSingleReadParsing _ =+    assertEqual "should parse single ACL property"+        (Just $ buildStreamACL $ setReadRole "iron")+        (decode $ fromStrict aclSingleReadBytes)++--------------------------------------------------------------------------------+aclMultipleReadParsing :: a -> IO ()+aclMultipleReadParsing _ =+    assertEqual "should parse multiple ACL property"+        (Just $ buildStreamACL $ setReadRoles ["iron", "game"])+        (decode $ fromStrict aclMultipleReadBytes)++--------------------------------------------------------------------------------+emptyStreamMetaParsing :: a -> IO ()+emptyStreamMetaParsing _ =+    assertEqual "should parse empty stream metadata"+        (Just emptyStreamMetadata)+        (decode $ fromStrict emptyMetaBytes)++--------------------------------------------------------------------------------+metaPropParsing :: a -> IO ()+metaPropParsing _ =+    assertEqual "should parse stream metadata property"+        (Just $ buildStreamMetadata $ setMaxCount 777)+        (decode $ fromStrict metaPropBytes)  -------------------------------------------------------------------------------- createPersistentTest :: Connection -> IO ()