diff --git a/CHANGELOG.markdown b/CHANGELOG.markdown --- a/CHANGELOG.markdown +++ b/CHANGELOG.markdown @@ -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. 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 @@ -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 () -------------------------------------------------------------------------------- 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 @@ -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 diff --git a/Database/EventStore/Internal/Manager/Operation/Registry.hs b/Database/EventStore/Internal/Manager/Operation/Registry.hs --- a/Database/EventStore/Internal/Manager/Operation/Registry.hs +++ b/Database/EventStore/Internal/Manager/Operation/Registry.hs @@ -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 diff --git a/Database/EventStore/Internal/Types.hs b/Database/EventStore/Internal/Types.hs --- a/Database/EventStore/Internal/Types.hs +++ b/Database/EventStore/Internal/Types.hs @@ -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. diff --git a/eventstore.cabal b/eventstore.cabal --- a/eventstore.cabal +++ b/eventstore.cabal @@ -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 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 diff --git a/tests/Test/Integration/Tests.hs b/tests/Test/Integration/Tests.hs --- a/tests/Test/Integration/Tests.hs +++ b/tests/Test/Integration/Tests.hs @@ -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 ()