diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,3 +1,8 @@ +# v0.5.2 + +* Fixed the undesirable behaviour that causes an empty response when `BySeqNum (-1)` is specified +* Expanded `Reconnect` so that the query that caused a reconnection is visible + # v0.5.1 * Fixed the bug adding all the following payloads in the result of `index` diff --git a/franz.cabal b/franz.cabal --- a/franz.cabal +++ b/franz.cabal @@ -1,12 +1,12 @@ cabal-version: 2.0 -version: 0.5.1 +version: 0.5.2 name: franz description: Please see the README on GitHub at homepage: https://github.com/tsurucapital/franz#readme bug-reports: https://github.com/tsurucapital/franz/issues author: Fumiaki Kinoshita maintainer: fumiexcel@gmail.com -copyright: Copyright (c) 2021 Fumiaki Kinoshita +copyright: Copyright (c) 2022 Fumiaki Kinoshita license: BSD3 license-file: LICENSE build-type: Simple diff --git a/src/Database/Franz/Client/Reconnect.hs b/src/Database/Franz/Client/Reconnect.hs --- a/src/Database/Franz/Client/Reconnect.hs +++ b/src/Database/Franz/Client/Reconnect.hs @@ -1,5 +1,6 @@ {-# LANGUAGE RankNTypes #-} {-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE LambdaCase #-} module Database.Franz.Client.Reconnect @@ -35,20 +36,23 @@ -> (STM Response -> IO r) -> IO r fetchWithPool pool q cont = withReconnection pool $ \conn -> fetch conn q cont + `catch` \case + r@(ReconnectInQuery _ _) -> throwM r -- Avoid deeply nested ReconnectInQuery + r -> throwM $ ReconnectInQuery q r -- | Run an action which takes a 'Connection', reconnecting whenever it throws an exception. withReconnection :: Pool -> (Connection -> IO a) -> IO a withReconnection Pool{..} cont = recovering poolRetryPolicy - [const $ Handler $ \Reconnect -> pure True] + [const $ Handler $ \(_ :: Reconnect) -> pure True] body where handler ex | Just (ClientError err) <- fromException ex = Just err | Just e <- fromException ex = Just (show (e :: IOException)) - | Just Reconnect <- fromException ex = Just - $ "Connection to " <> fromFranzPath poolPath <> " timed out" + | Just (e :: Reconnect) <- fromException ex = Just + $ "Reconnecting to " <> fromFranzPath poolPath <> " due to " <> show e | otherwise = Nothing body _ = do @@ -59,7 +63,7 @@ , fromFranzPath poolPath ] conn <- tryJust handler (connect poolPath) - >>= either (\e -> poolLogFunc e >> throwM Reconnect) pure + >>= either (\e -> poolLogFunc e >> throwM ReconnectByError) pure poolLogFunc $ "Connection #" <> show i <> " established" pure ((i, Just conn), (i, conn)) v@(i, Just c) -> pure (v, (i, c)) @@ -73,9 +77,12 @@ -- another thread already established a new connection (j, Just _) | i == j -> (i + 1, Nothing) <$ disconnect conn x -> pure x - throwM Reconnect + throwM ReconnectByError -data Reconnect = Reconnect deriving (Show, Eq) +data Reconnect = ReconnectByTimeout + | ReconnectByError + | ReconnectInQuery !Query !Reconnect + deriving (Show, Eq) instance Exception Reconnect withPool :: RetryPolicyM IO @@ -93,4 +100,4 @@ atomicallyReconnecting :: Int -- ^ timeout in microseconds -> STM a -> IO a atomicallyReconnecting timeout m = atomicallyWithin timeout m - >>= maybe (throwM Reconnect) pure + >>= maybe (throwM ReconnectByTimeout) pure diff --git a/src/Database/Franz/Internal/Protocol.hs b/src/Database/Franz/Internal/Protocol.hs --- a/src/Database/Franz/Internal/Protocol.hs +++ b/src/Database/Franz/Internal/Protocol.hs @@ -59,12 +59,12 @@ data RequestType = AllItems | LastItem | FirstItem - deriving (Show, Generic) + deriving (Show, Eq, Generic) instance Serialize RequestType data ItemRef = BySeqNum !Int -- ^ sequential number | ByIndex !IndexName !Int -- ^ index name and value - deriving (Show, Generic) + deriving (Show, Eq, Generic) instance Serialize ItemRef data Query = Query @@ -72,7 +72,7 @@ , reqFrom :: !ItemRef -- ^ name of the index to search , reqTo :: !ItemRef -- ^ name of the index to search , reqType :: !RequestType - } deriving (Show, Generic) + } deriving (Show, Eq, Generic) instance Serialize Query data RawRequest diff --git a/src/Database/Franz/Internal/Reader.hs b/src/Database/Franz/Internal/Reader.hs --- a/src/Database/Franz/Internal/Reader.hs +++ b/src/Database/Franz/Internal/Reader.hs @@ -136,23 +136,26 @@ type QueryResult = ((Int, Int) -- SeqNo *before* the first result, byte offset , (Int, Int)) -- SeqNo of the last result, byte offset +isEmptyResult :: QueryResult -> Bool +isEmptyResult ((i, _), (j, _)) = i >= j + range :: Int -- ^ from -> Int -- ^ to -> RequestType -> IM.IntMap Int -- ^ offsets - -> Maybe QueryResult -range begin end rt allOffsets = case rt of - AllItems -> (firstItem, maybe firstItem fst $ IM.maxViewWithKey body) <$ guard ready + -> (Bool, QueryResult) +range begin end rt allOffsets = (,) ready $ case rt of + AllItems -> (firstItem, maybe firstItem fst $ IM.maxViewWithKey body) LastItem -> case IM.maxViewWithKey body of - Nothing -> (zero, zero) <$ guard ready + Nothing -> (zero, zero) Just (ofs', r) -> case IM.maxViewWithKey (IM.union left r) of - Just (ofs, _) -> (ofs, ofs') <$ guard ready - Nothing -> (zero, ofs') <$ guard ready + Just (ofs, _) -> (ofs, ofs') + Nothing -> (zero, ofs') FirstItem -> case IM.minViewWithKey body of - Nothing -> (zero, zero) <$ guard ready + Nothing -> (zero, zero) Just (ofs', _) -> case IM.maxViewWithKey left of - Just (ofs, _) -> (ofs, ofs') <$ guard ready - Nothing -> (zero, ofs') <$ guard ready + Just (ofs, _) -> (ofs, ofs') + Nothing -> (zero, ofs') where ------------------------------------ -- begin end @@ -325,7 +328,17 @@ let (body, lastItem, _) = IM.splitLookup val m let body' = maybe id (IM.insert val) lastItem body return $! maybe minBound fst $ IM.maxView body' - return $! range begin end rt allOffsets + return $! case range begin end rt allOffsets of + (ready, result) + -- When BySeqNum (-1) is specified, the client expects at least one item. + -- More concretely, (ByIndex "time" t, BySeqNum (-1)) should resolve ByIndex + -- until there's an item with a timestamp later than t. + -- Perhaps there should be a special constructor for ItemRef that points to + -- the latest item at transaction, and redefine "Nth-from-end" semantics + -- where the "end" is frozen at the reception of the query. + | BySeqNum i <- end_, i < 0, isEmptyResult result -> Nothing + | ready -> Just result + | otherwise -> Nothing where acquire = join $ atomically $ do allStreams <- readTVar vStreams diff --git a/tests/test.hs b/tests/test.hs --- a/tests/test.hs +++ b/tests/test.hs @@ -12,42 +12,43 @@ main :: IO () main = withSystemTempDirectory "franz-test" $ \path -> do - withWriter (Identity "") path - $ \h -> forM_ ([0..9] :: [Int]) - $ \i -> write h (Identity (fromIntegral i * 10)) (BB.intDec i) + withWriter (Identity "") path + $ \h -> forM_ ([0..9] :: [Int]) + $ \i -> write h (Identity (fromIntegral i * 10)) (BB.intDec i) - numFails <- newIORef (0 :: Int) + numFails <- newIORef (0 :: Int) - withConnection (LocalFranzPath path) $ \conn -> do - let test :: RequestType -> ItemRef -> ItemRef -> [Item] -> IO () - test ty i j expected = do - putStr $ unwords ["test", show ty, showsPrec 11 i "", showsPrec 11 j "", show expected, ": "] - fetchSimple conn 500000 (Query "" i j ty) >>= \case - Nothing -> error "timeout" - Just xs - | C.toList xs == expected -> putStrLn "ok" - | otherwise -> do - modifyIORef' numFails (+1) - putStrLn $ "got " <> show (C.toList xs) + withConnection (LocalFranzPath path) $ \conn -> do + let test :: RequestType -> ItemRef -> ItemRef -> Maybe [Item] -> IO () + test ty i j expected = do + putStr $ unwords ["test", show ty, showsPrec 11 i "", showsPrec 11 j "", show expected, ": "] + result <- fmap C.toList <$> fetchSimple conn 100000 (Query "" i j ty) + if result == expected + then putStrLn "ok" + else do + modifyIORef' numFails (+1) + putStrLn $ "got " <> show result - test AllItems (BySeqNum 1) (BySeqNum 3) [Item {seqNo = 1, indices = [10], payload = "1"},Item {seqNo = 2, indices = [20], payload = "2"},Item {seqNo = 3, indices = [30], payload = "3"}] - test AllItems (ByIndex "" 15) (ByIndex "" 15) [] - test AllItems (ByIndex "" 15) (ByIndex "" 25) [Item {seqNo = 2, indices = [20], payload = "2"}] - test AllItems (ByIndex "" 10) (ByIndex "" 20) [Item {seqNo = 1, indices = [10], payload = "1"},Item {seqNo = 2, indices = [20], payload = "2"}] - test AllItems (ByIndex "" 15) (ByIndex "" 20) [Item {seqNo = 2, indices = [20], payload = "2"}] + test AllItems (BySeqNum 1) (BySeqNum 3) $ Just [Item {seqNo = 1, indices = [10], payload = "1"},Item {seqNo = 2, indices = [20], payload = "2"},Item {seqNo = 3, indices = [30], payload = "3"}] + test AllItems (ByIndex "" 15) (ByIndex "" 15) $ Just [] + test AllItems (ByIndex "" 15) (ByIndex "" 25) $ Just [Item {seqNo = 2, indices = [20], payload = "2"}] + test AllItems (ByIndex "" 10) (ByIndex "" 20) $ Just [Item {seqNo = 1, indices = [10], payload = "1"},Item {seqNo = 2, indices = [20], payload = "2"}] + test AllItems (ByIndex "" 15) (ByIndex "" 20) $ Just [Item {seqNo = 2, indices = [20], payload = "2"}] - test FirstItem (BySeqNum 1) (BySeqNum 5) [Item {seqNo = 1, indices = [10], payload = "1"}] - test FirstItem (ByIndex "" 10) (ByIndex "" 45) [Item {seqNo = 1, indices = [10], payload = "1"}] - test FirstItem (ByIndex "" 15) (ByIndex "" 45) [Item {seqNo = 2, indices = [20], payload = "2"}] - test FirstItem (ByIndex "" 10) (ByIndex "" 10) [Item {seqNo = 1, indices = [10], payload = "1"}] - test FirstItem (ByIndex "" 15) (ByIndex "" 15) [] + test FirstItem (BySeqNum 1) (BySeqNum 5) $ Just [Item {seqNo = 1, indices = [10], payload = "1"}] + test FirstItem (ByIndex "" 10) (ByIndex "" 45) $ Just [Item {seqNo = 1, indices = [10], payload = "1"}] + test FirstItem (ByIndex "" 15) (ByIndex "" 45) $ Just [Item {seqNo = 2, indices = [20], payload = "2"}] + test FirstItem (ByIndex "" 10) (ByIndex "" 10) $ Just [Item {seqNo = 1, indices = [10], payload = "1"}] + test FirstItem (ByIndex "" 15) (ByIndex "" 15) $ Just [] - test LastItem (BySeqNum 1) (BySeqNum 5) [Item {seqNo = 5, indices = [50], payload = "5"}] - test LastItem (ByIndex "" 10) (ByIndex "" 45) [Item {seqNo = 4, indices = [40], payload = "4"}] - test LastItem (ByIndex "" 15) (ByIndex "" 50) [Item {seqNo = 5, indices = [50], payload = "5"}] - test LastItem (ByIndex "" 10) (ByIndex "" 10) [Item {seqNo = 1, indices = [10], payload = "1"}] - test LastItem (ByIndex "" 15) (ByIndex "" 15) [] + test LastItem (BySeqNum 1) (BySeqNum 5) $ Just [Item {seqNo = 5, indices = [50], payload = "5"}] + test LastItem (ByIndex "" 10) (ByIndex "" 45) $ Just [Item {seqNo = 4, indices = [40], payload = "4"}] + test LastItem (ByIndex "" 15) (ByIndex "" 50) $ Just [Item {seqNo = 5, indices = [50], payload = "5"}] + test LastItem (ByIndex "" 10) (ByIndex "" 10) $ Just [Item {seqNo = 1, indices = [10], payload = "1"}] + test LastItem (ByIndex "" 15) (ByIndex "" 15) $ Just [] - readIORef numFails >>= \case - 0 -> putStrLn "All good" - _ -> fail "Failed" + test LastItem (BySeqNum 10) (BySeqNum (-1)) Nothing + + readIORef numFails >>= \case + 0 -> putStrLn "All good" + _ -> fail "Failed"