diff --git a/grapesy.cabal b/grapesy.cabal index e42f773d..7796bfb7 100644 --- a/grapesy.cabal +++ b/grapesy.cabal @@ -309,14 +309,18 @@ test-suite test-grapesy Test.Prop.IncrementalParsing Test.Prop.Serialization Test.Sanity.BrokenDeployments + Test.Sanity.Disconnect Test.Sanity.EndOfStream + Test.Sanity.Exception Test.Sanity.Interop Test.Sanity.StreamingType.CustomFormat Test.Sanity.StreamingType.NonStreaming Test.Util Test.Util.Awkward + Test.Util.Exception Test.Util.Orphans Test.Util.Protobuf + Test.Util.RawTestServer -- Internals we're testing Network.GRPC.Util.Parser @@ -342,6 +346,7 @@ test-suite test-grapesy , bytestring >= 0.10 && < 0.13 , case-insensitive >= 1.2 && < 1.3 , containers >= 0.6 && < 0.8 + , directory >= 1.3 && < 1.4 , exceptions >= 0.10 && < 0.11 , http-types >= 0.12 && < 0.13 , http2 >= 5.3.1 && < 5.4 @@ -363,6 +368,7 @@ test-suite test-grapesy , text >= 1.2 && < 2.2 , tls >= 1.7 && < 2.2 , tree-diff >= 0.3 && < 0.4 + , unix >= 2.7 && < 2.9 , utf8-string >= 1.0 && < 1.1 executable demo-client diff --git a/interop/Interop/Client/TestCase/CustomMetadata.hs b/interop/Interop/Client/TestCase/CustomMetadata.hs index a567298f..20daf465 100644 --- a/interop/Interop/Client/TestCase/CustomMetadata.hs +++ b/interop/Interop/Client/TestCase/CustomMetadata.hs @@ -21,7 +21,7 @@ import Proto.API.Interop -- For both UnaryCall and FullDuplexCall, the reference server (at least some) -- does not return any initial metadata until we send the first request. The -- test spec does not specify whether this is expected behaviour or not, so we --- play it save and only ask for the initial metadata after sending the request. +-- play it safe and only ask for the initial metadata after sending the request. runTest :: Cmdline -> IO () runTest cmdline = do withConnection def (testServer cmdline) $ \conn -> do diff --git a/src/Network/GRPC/Server/Call.hs b/src/Network/GRPC/Server/Call.hs index bbbdbbfc..4f7f4113 100644 --- a/src/Network/GRPC/Server/Call.hs +++ b/src/Network/GRPC/Server/Call.hs @@ -628,7 +628,7 @@ recvEndOfInput call@Call{} = do -- | Send 'ProperTrailers' -- --- This function is not part of the public API: we use it the top-level +-- This function is not part of the public API: we use it as the top-level -- exception handler in "Network.GRPC.Server" to forward exceptions in server -- handlers to the client. -- diff --git a/src/Network/GRPC/Server/Handler.hs b/src/Network/GRPC/Server/Handler.hs index 8aeaea15..f323bc1c 100644 --- a/src/Network/GRPC/Server/Handler.hs +++ b/src/Network/GRPC/Server/Handler.hs @@ -272,7 +272,7 @@ waitForHandler unmask call handlerThread = loop -- -- The attempt to forward it to the client is a best-effort only: -- --- * The nature of the exception might mean that we we cannot send anything to +-- * The nature of the exception might mean that we cannot send anything to -- the client at all. -- * It is possible the exception was thrown /after/ the handler already send -- the trailers to the client. diff --git a/test-grapesy/Main.hs b/test-grapesy/Main.hs index a3575150..a62daf31 100644 --- a/test-grapesy/Main.hs +++ b/test-grapesy/Main.hs @@ -17,7 +17,9 @@ import Test.Prop.Dialogue qualified as Dialogue import Test.Prop.IncrementalParsing qualified as IncrementalParsing import Test.Prop.Serialization qualified as Serialization import Test.Sanity.BrokenDeployments qualified as BrokenDeployments +import Test.Sanity.Disconnect qualified as Disconnect import Test.Sanity.EndOfStream qualified as EndOfStream +import Test.Sanity.Exception qualified as Exception import Test.Sanity.Interop qualified as Interop import Test.Sanity.StreamingType.CustomFormat qualified as StreamingType.CustomFormat import Test.Sanity.StreamingType.NonStreaming qualified as StreamingType.NonStreaming @@ -28,11 +30,13 @@ main = do defaultMain $ testGroup "grapesy" [ testGroup "Sanity" [ - EndOfStream.tests + Disconnect.tests + , EndOfStream.tests , testGroup "StreamingType" [ StreamingType.NonStreaming.tests , StreamingType.CustomFormat.tests ] + , Exception.tests , Interop.tests , BrokenDeployments.tests ] diff --git a/test-grapesy/Test/Driver/ClientServer.hs b/test-grapesy/Test/Driver/ClientServer.hs index d9d70bb4..2678a31a 100644 --- a/test-grapesy/Test/Driver/ClientServer.hs +++ b/test-grapesy/Test/Driver/ClientServer.hs @@ -40,6 +40,7 @@ import Network.GRPC.Common import Network.GRPC.Common.Compression qualified as Compr import Network.GRPC.Server qualified as Server import Network.GRPC.Server.Run qualified as Server +import Test.Util.Exception import Paths_grapesy @@ -168,12 +169,6 @@ data TlsFail = we don't see these exceptions server-side. -------------------------------------------------------------------------------} --- | Exception thrown by client or handler to test exception handling -data DeliberateException = forall e. Exception e => DeliberateException e - deriving anyclass (Exception) - -deriving stock instance Show DeliberateException - isExpectedServerException :: ClientServerConfig -> SomeException -> Bool isExpectedServerException cfg e -- @@ -232,7 +227,7 @@ isExpectedClientException cfg e | Just (DeliberateException _) <- fromException e = True - -- Server threw deliberat exception + -- Server threw deliberate exception | Just grpcException <- fromException e , Just msg <- grpcErrorMessage grpcException , "DeliberateException" `Text.isInfixOf` msg diff --git a/test-grapesy/Test/Driver/Dialogue/Definition.hs b/test-grapesy/Test/Driver/Dialogue/Definition.hs index 802de12d..53541d04 100644 --- a/test-grapesy/Test/Driver/Dialogue/Definition.hs +++ b/test-grapesy/Test/Driver/Dialogue/Definition.hs @@ -20,7 +20,6 @@ module Test.Driver.Dialogue.Definition ( , hasEarlyTermination ) where -import Control.Exception import Control.Monad.State (StateT, execStateT, modify) import Data.Bifunctor import Data.ByteString qualified as Strict (ByteString) @@ -28,6 +27,7 @@ import Data.ByteString qualified as Strict (ByteString) import Network.GRPC.Common import Test.Driver.Dialogue.TestClock qualified as TestClock +import Test.Util.Exception import Control.Monad.Catch import GHC.Show (appPrec1, showCommaSpace) @@ -154,25 +154,6 @@ newtype GlobalSteps = GlobalSteps { } deriving stock (Show) -{------------------------------------------------------------------------------- - User exceptions - - When a test calls for the client or the server to throw an exception, we throw - one of these. Their sole purpose is to be "any" kind of exception (not a - specific one). --------------------------------------------------------------------------------} - -data SomeServerException = SomeServerException ExceptionId - deriving stock (Show, Eq) - deriving anyclass (Exception) - -data SomeClientException = SomeClientException ExceptionId - deriving stock (Show, Eq) - deriving anyclass (Exception) - --- | We distinguish exceptions from each other simply by a number -type ExceptionId = Int - {------------------------------------------------------------------------------- Utility -------------------------------------------------------------------------------} diff --git a/test-grapesy/Test/Driver/Dialogue/Execution.hs b/test-grapesy/Test/Driver/Dialogue/Execution.hs index 25f2ba03..13eea7bb 100644 --- a/test-grapesy/Test/Driver/Dialogue/Execution.hs +++ b/test-grapesy/Test/Driver/Dialogue/Execution.hs @@ -413,6 +413,8 @@ serverLocal clock call = \(LocalSteps steps) -> do Terminate mErr -> do mInp <- liftIO $ try $ within timeoutReceive action $ Server.Binary.recvInput call + -- TODO: + -- -- On the server side we cannot distinguish regular client -- termination from an exception when receiving. let expectation = isExpectedElem $ NoMoreElems NoMetadata @@ -426,6 +428,12 @@ serverLocal clock call = \(LocalSteps steps) -> do -- terminate more-or-less immediately, this does not necessarily indicate -- any kind of failure: the client may simply have put the call in -- half-closed mode. + -- + -- TODO: + -- However, when the client terminates early and we are not using one + -- connection per RPC (i.e. we are sharing a connection), the server will + -- /never/ realize that the client has disappeared. See the discussion in + -- the issue above. waitForClientDisconnect :: IO () waitForClientDisconnect = within timeoutFailure () $ loop diff --git a/test-grapesy/Test/Sanity/BrokenDeployments.hs b/test-grapesy/Test/Sanity/BrokenDeployments.hs index a36bb613..ec0ea9a1 100644 --- a/test-grapesy/Test/Sanity/BrokenDeployments.hs +++ b/test-grapesy/Test/Sanity/BrokenDeployments.hs @@ -3,26 +3,18 @@ module Test.Sanity.BrokenDeployments (tests) where -import Control.Concurrent -import Control.Concurrent.Async import Control.Exception -import Data.ByteString qualified as BS.Strict -import Data.ByteString qualified as Strict (ByteString) -import Data.ByteString.Builder qualified as BS.Builder import Data.ByteString.Char8 qualified as BS.Strict.Char8 import Data.ByteString.UTF8 qualified as BS.Strict.UTF8 -import Data.String (fromString) import Data.Text qualified as Text import Network.HTTP.Types qualified as HTTP -import Network.HTTP2.Server qualified as HTTP2 -import Network.Run.TCP qualified as NetworkRun -import Network.Socket import Test.Tasty import Test.Tasty.HUnit import Network.GRPC.Client qualified as Client import Network.GRPC.Common import Network.GRPC.Common.Protobuf +import Test.Util.RawTestServer import Proto.API.Ping @@ -54,6 +46,11 @@ tests = testGroup "Test.Sanity.BrokenDeployments" [ ] ] +connParams :: Client.ConnParams +connParams = def { + Client.connVerifyHeaders = True + } + {------------------------------------------------------------------------------- HTTP Status -------------------------------------------------------------------------------} @@ -322,101 +319,6 @@ test_invalidTrailerMetadata = respondWith response $ \addr -> do someInvalidMetadata :: String someInvalidMetadata = "This is invalid: 你好" -{------------------------------------------------------------------------------- - Test server - - This allows us to simulate broken /servers/. --------------------------------------------------------------------------------} - -data Response = Response { - responseStatus :: HTTP.Status - , responseHeaders :: [HTTP.Header] - , responseBody :: Strict.ByteString - , responseTrailers :: [HTTP.Header] - } - -instance Default Response where - def = Response { - responseStatus = HTTP.ok200 - , responseHeaders = [ asciiHeader "content-type" "application/grpc" ] - , responseBody = BS.Strict.empty - , responseTrailers = [ asciiHeader "grpc-status" "0" ] - } - --- | Server that responds with the given 'Response', independent of the request -respondWith :: Response -> (Client.Address -> IO a) -> IO a -respondWith response = withTestServer $ \_req _aux respond -> - respond http2Response [] - where - http2Response :: HTTP2.Response - http2Response = - flip HTTP2.setResponseTrailersMaker trailersMaker $ - HTTP2.responseBuilder - (responseStatus response) - (responseHeaders response) - (BS.Builder.byteString $ responseBody response) - - trailersMaker :: HTTP2.TrailersMaker - trailersMaker Nothing = return $ HTTP2.Trailers (responseTrailers response) - trailersMaker (Just _) = return $ HTTP2.NextTrailersMaker trailersMaker - --- | Low-level test server --- --- We bypass the entire grapesy machinery for constructing the server, because --- we need to mock a broken deployment. --- --- The grapesy client can auto reconnect when the server is not (yet) up and --- running, but to keep things simple, and since the server anyway runs in the --- same process, we just signal when the server is ready. This also allows us --- to avoid binding to a specific port in the tests (which might already be in --- use on the machine running the tests, leading to spurious test failures). -testServer :: HTTP2.Server -> MVar PortNumber -> IO () -testServer server serverPort = do - addr <- NetworkRun.resolve Stream (Just "127.0.0.1") "0" [AI_PASSIVE] - bracket (NetworkRun.openTCPServerSocket addr) close $ \listenSock -> do - addr' <- getSocketName listenSock - port <- case addr' of - SockAddrInet port _host -> return port - SockAddrInet6 port _ _host _ -> return port - SockAddrUnix{} -> error "respondWith: unexpected unix socket" - putMVar serverPort port - NetworkRun.runTCPServerWithSocket listenSock $ \clientSock -> - bracket (HTTP2.allocSimpleConfig clientSock 4096) - HTTP2.freeSimpleConfig $ \config -> - HTTP2.run HTTP2.defaultServerConfig config server - -withTestServer :: HTTP2.Server -> (Client.Address -> IO a) -> IO a -withTestServer server k = do - serverPort <- newEmptyMVar - withAsync (testServer server serverPort) $ \_serverThread -> do - port <- readMVar serverPort - let addr :: Client.Address - addr = Client.Address { - addressHost = "127.0.0.1" - , addressPort = port - , addressAuthority = Nothing - } - k addr - -{------------------------------------------------------------------------------- - Auxiliary --------------------------------------------------------------------------------} - -connParams :: Client.ConnParams -connParams = def { - Client.connVerifyHeaders = True - } - --- | Header with ASCII value --- --- (Header /names/ are always ASCII.) -asciiHeader :: String -> String -> HTTP.Header -asciiHeader name value = (fromString name, BS.Strict.Char8.pack value) - --- | Header with UTF-8 encoded value -utf8Header :: String -> String -> HTTP.Header -utf8Header name value = (fromString name, BS.Strict.UTF8.fromString value) - grpcMessageContains :: GrpcException -> String -> Bool grpcMessageContains GrpcException{grpcErrorMessage} str = case grpcErrorMessage of diff --git a/test-grapesy/Test/Sanity/Disconnect.hs b/test-grapesy/Test/Sanity/Disconnect.hs new file mode 100644 index 00000000..2d164a0d --- /dev/null +++ b/test-grapesy/Test/Sanity/Disconnect.hs @@ -0,0 +1,288 @@ +{-# OPTIONS_GHC -Wno-orphans #-} + +-- | Handling of client or server disconnections occurring with ongoing RPCs on +-- a shared connection. +-- +-- When a server disconnects, we expect: +-- +-- 1. All current calls fail with 'Client.ServerDisconnected' +-- 2. Future calls (after reconnection) succeed +-- +-- When a client disconnects, we expect: +-- +-- 1. The handlers dealing with that client (i.e. on that connection) should +-- fail with 'Server.ClientDisonnected' +-- 2. Future calls (after reconnection) succeed +module Test.Sanity.Disconnect where + +import Control.Concurrent +import Control.Concurrent.Async +import Control.Exception +import Control.Monad +import Data.ByteString.Lazy qualified as Lazy (ByteString) +import Data.Either +import Data.IORef +import Data.Word +import Foreign.C.Types (CInt(..)) +import System.Posix +import Test.Tasty +import Test.Tasty.HUnit +import Text.Read + +import Network.GRPC.Client qualified as Client +import Network.GRPC.Client.Binary qualified as Binary +import Network.GRPC.Common +import Network.GRPC.Server qualified as Server +import Network.GRPC.Server.Binary qualified as Binary +import Network.GRPC.Spec +import Test.Util +import Test.Util.RawTestServer + +tests :: TestTree +tests = testGroup "Test.Sanity.Disconnect" [ + testCase "client" test_clientDisconnect + , testCase "server" test_serverDisconnect + ] + +-- | Two separate clients make many concurrent calls, one of them disconnects. +test_clientDisconnect :: Assertion +test_clientDisconnect = do + -- Create the server + disconnectCounter1 <- newIORef 0 + disconnectCounter2 <- newIORef 0 + server <- + Server.mkGrpcServer def [ + Server.someRpcHandler $ + Server.mkRpcHandler @Trivial $ echoHandler (Just disconnectCounter1) + , Server.someRpcHandler $ + Server.mkRpcHandler @Trivial' $ echoHandler (Just disconnectCounter2) + ] + + portSignal <- newEmptyMVar + void $ forkIO $ rawTestServer (pure Nothing) (putMVar portSignal) server + + -- Start server + serverPort <- readMVar portSignal + let serverAddress = + Client.ServerInsecure Client.Address { + addressHost = "127.0.0.1" + , addressPort = serverPort + , addressAuthority = Nothing + } + + -- Start a client in a separate process + void $ forkProcess $ + Client.withConnection def serverAddress $ \conn -> do + -- Make 50 concurrent calls. 49 of them sending infinite messages. One + -- of them kills this client process after 100 messages. + let numCalls = 50 + predicate = pure . const False + predicates = + replicate (numCalls - 1) predicate ++ + [ \n -> do + when (n == 100) $ c_exit 1 + return False + ] + mapConcurrently_ + ( Client.withRPC conn def (Proxy @Trivial) + . countUntil + ) + predicates + + -- Start two more clients that make 50 calls to each handler, all calls + -- counting up to 1000 + let numCalls = 50 + countTo = 100 + predicate = pure . (>= countTo) + predicates = replicate numCalls predicate + (result1, result2) <- concurrently + ( Client.withConnection def serverAddress $ \conn -> do + sum <$> mapConcurrently + ( Client.withRPC conn def (Proxy @Trivial) + . countUntil + ) + predicates + ) + ( Client.withConnection def serverAddress $ \conn -> do + sum <$> mapConcurrently + ( Client.withRPC conn def (Proxy @Trivial') + . countUntil + ) + predicates + ) + + -- All calls should have finished with a results of 'countTo', for both + -- clients + assertBool "" (result1 + result2 == 2 * sum (replicate numCalls countTo)) + + -- We should also see only 50 client disconnects for the first handler and + -- none for the second + clientDisconnects1 <- readIORef disconnectCounter1 + clientDisconnects2 <- readIORef disconnectCounter2 + assertBool "" (clientDisconnects1 == 50 && clientDisconnects2 == 0) + +-- | Client makes many concurrent calls, server disconnects +test_serverDisconnect :: Assertion +test_serverDisconnect = withTemporaryFile $ \ipcFile -> do + -- We use a temporary file as a very rudimentary means of inter-process + -- communication so the server (which runs in a separate process) can make + -- the client aware of the port it is assigned by the OS. This also helps us + -- make sure the server binds to the same port when it comes back up for + -- reconnect purposes. + let ipcWrite :: String -> IO () + ipcWrite msg = do + writeFile ipcFile "" + writeFile ipcFile msg + + ipcRead :: IO String + ipcRead = readFile ipcFile + + ipcWaitRead :: IO String + ipcWaitRead = do + ipcRead >>= \case + "" -> do + threadDelay 10000 >> ipcWaitRead + msg -> do + return msg + + -- Create the server + server <- + Server.mkGrpcServer def [ + Server.someRpcHandler $ + Server.mkRpcHandler @Trivial $ echoHandler Nothing + ] + + let -- Starts the server in a new process. Gives back an action that kills + -- the server process. + startServer :: IO (IO ()) + startServer = do + serverPid <- + forkProcess $ + rawTestServer (readMaybe <$> ipcRead) (ipcWrite . show) server + return $ c_kill (fromIntegral serverPid) sigKILL + + -- Start server, get the port + killServer <- startServer + serverPort <- read <$> ipcWaitRead + signalRestart <- newEmptyMVar + let serverAddress = + Client.ServerInsecure Client.Address { + addressHost = "127.0.0.1" + , addressPort = serverPort + , addressAuthority = Nothing + } + + reconnectPolicy :: Client.ReconnectPolicy + reconnectPolicy = go 0 + where + go :: Int -> Client.ReconnectPolicy + go n + | n == 5 + = Client.ReconnectAfter $ do + killRestarted <- startServer + putMVar signalRestart killRestarted + return $ Client.exponentialBackoff threadDelay 1 (1, 1) 100 + | otherwise + = Client.ReconnectAfter $ do + threadDelay 10000 + return $ go (n + 1) + + connParams :: Client.ConnParams + connParams = def { Client.connReconnectPolicy = reconnectPolicy } + + Client.withConnection connParams serverAddress $ \conn -> do + -- Make 50 concurrent calls. 49 of them sending infinite messages. One + -- of them kills the server after 100 messages. + let numCalls = 50 + predicate = pure . const False + predicates = + replicate (numCalls - 1) predicate ++ + [ \n -> do + when (n == 100) killServer + return False + ] + results <- + mapConcurrently + ( try @Client.ServerDisconnected + . Client.withRPC conn def (Proxy @Trivial) + . countUntil + ) + predicates + + -- All calls should have failed + assertBool "" (null (rights results) && length (lefts results) == numCalls) + + -- New calls should succeed (after reconnection) + killRestarted <- takeMVar signalRestart + result <- Client.withRPC conn def (Proxy @Trivial) $ + countUntil (pure . (>= 100)) + assertBool "" (result == 100) + + -- Do not leave the server process hanging around + killRestarted + +{------------------------------------------------------------------------------- + Client and handler functions +-------------------------------------------------------------------------------} + +-- | Send increasing numbers to the server until it responds with one that +-- satisfies the given predicate. +countUntil :: forall rpc. + ( Input rpc ~ Lazy.ByteString + , Output rpc ~ Lazy.ByteString + , ResponseTrailingMetadata rpc ~ NoMetadata + ) => (Word64 -> IO Bool) -> Client.Call rpc -> IO Word64 +countUntil = go 0 + where + go :: Word64 -> (Word64 -> IO Bool) -> Client.Call rpc -> IO Word64 + go next p call = do + sat <- p next + if sat then do + Binary.sendFinalInput @Word64 call next + (final, NoMetadata) <- Binary.recvFinalOutput @Word64 call + return final + else do + Binary.sendNextInput @Word64 call next + next' <- Binary.recvNextOutput @Word64 call + go (succ next') p call + +-- | Echos any input +echoHandler :: + ( Input rpc ~ Lazy.ByteString + , Output rpc ~ Lazy.ByteString + , ResponseTrailingMetadata rpc ~ NoMetadata + ) => Maybe (IORef Int) -> Server.Call rpc -> IO () +echoHandler disconnectCounter call = trackDisconnects disconnectCounter $ do + Binary.recvInput @Word64 call >>= \case + StreamElem n -> do + Binary.sendNextOutput @Word64 call n + echoHandler disconnectCounter call + FinalElem n _ -> do + Binary.sendFinalOutput @Word64 call (n, NoMetadata) + NoMoreElems _ -> do + Server.sendTrailers call NoMetadata + where + trackDisconnects Nothing = + id + trackDisconnects (Just counter) = + handle ( + \(_e :: Server.ClientDisconnected) -> + atomicModifyIORef' counter $ \n -> (n + 1, ()) + ) + +{------------------------------------------------------------------------------- + Auxiliary +-------------------------------------------------------------------------------} + +foreign import ccall unsafe "kill" c_kill :: CInt -> CInt -> IO () +foreign import ccall unsafe "exit" c_exit :: CInt -> IO () + +type Trivial = RawRpc "trivial" "trivial" +type Trivial' = RawRpc "trivial" "trivial'" + +type instance RequestMetadata Trivial = NoMetadata +type instance ResponseInitialMetadata Trivial = NoMetadata +type instance ResponseTrailingMetadata Trivial = NoMetadata +type instance RequestMetadata Trivial' = NoMetadata +type instance ResponseInitialMetadata Trivial' = NoMetadata +type instance ResponseTrailingMetadata Trivial' = NoMetadata diff --git a/test-grapesy/Test/Sanity/Exception.hs b/test-grapesy/Test/Sanity/Exception.hs new file mode 100644 index 00000000..e3312e74 --- /dev/null +++ b/test-grapesy/Test/Sanity/Exception.hs @@ -0,0 +1,196 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# OPTIONS_GHC -Wno-orphans #-} + +-- | Handling of exceptions occurring in an RPC on a shared connection. +-- +-- These tests check for the behavior described in +-- . In particular, there are +-- two conditions that should hold when an exception occurs in the scope of a +-- call (on either the server or client, e.g. inside either 'mkRpcHandler' or +-- 'withRPC'): +-- +-- 1. Other ongoing calls on that connection are not terminated, and +-- 2. future calls are still possible. +module Test.Sanity.Exception where + +import Control.Concurrent.Async +import Control.Exception +import Control.Monad +import Data.Either +import Data.IORef +import Data.Text qualified as Text +import Data.Word +import Test.Tasty +import Test.Tasty.HUnit + +import Network.GRPC.Client qualified as Client +import Network.GRPC.Client.Binary qualified as Binary +import Network.GRPC.Common +import Network.GRPC.Server qualified as Server +import Network.GRPC.Server.Binary qualified as Binary +import Network.GRPC.Spec +import Test.Driver.ClientServer +import Test.Util.Exception + +tests :: TestTree +tests = testGroup "Test.Sanity.Exception" [ + testCase "client" test_clientException + , testCase "server" test_serverException + + , testCase "earlyTerminationNoWait" test_earlyTerminationNoWait + ] + +-- | Client makes many concurrent calls, throws an exception during one of them. +test_clientException :: IO () +test_clientException = testClientServer $ ClientServerTest { + config = def + , client = simpleTestClient $ \conn -> do + -- Make 100 concurrent calls. 99 of them counting to 50, and one + -- more that throws an exception once it reaches 10. + let + predicate = (> 50) + predicates = + replicate 99 predicate ++ + [ \n -> + (n > 10) + && throw (DeliberateException $ SomeClientException 1) + ] + + results <- + mapConcurrently + ( try @DeliberateException + . Client.withRPC conn def (Proxy @Trivial) + . countUntil + ) + predicates + + -- Only one of the calls failed + assertEqual "" (length $ lefts results) 1 + + -- All others terminated with results satisfying the predicate + assertBool "" (all predicate $ rights results) + + -- New calls still succeed + assertBool "" . predicate + =<< Client.withRPC conn def (Proxy @Trivial) (countUntil predicate) + , server = [ + Server.someRpcHandler $ + Server.mkRpcHandler @Trivial incUntilFinal + ] + } + +-- | Client makes many concurrent calls, the handler throws an exception during +-- one of them. +test_serverException :: IO () +test_serverException = do + handlerCounter <- newIORef @Int 0 + testClientServer $ ClientServerTest { + config = def { expectEarlyServerTermination = True } + , client = simpleTestClient $ \conn -> do + -- Make 100 concurrent calls counting to 50. + let predicate = (> 50) + results <- + replicateConcurrently 100 $ + try @GrpcException + . Client.withRPC conn def (Proxy @Trivial) + $ countUntil predicate + + -- Only one of the calls failed, and we got the appropriate + -- exception + case lefts results of + [GrpcException GrpcUnknown (Just msg) []] -> do + assertBool "" $ "DeliberateException" `Text.isInfixOf` msg + assertBool "" $ "SomeServerException 1" `Text.isInfixOf` msg + _ -> + assertFailure "" + + -- All others terminated with results satisfying the predicate + assertBool "" (all predicate $ rights results) + + -- New calls still succeed + assertBool "" . predicate + =<< Client.withRPC conn def (Proxy @Trivial) (countUntil predicate) + , server = [ + Server.someRpcHandler $ + Server.mkRpcHandler @Trivial $ \call -> do + handlerCount <- + atomicModifyIORef' handlerCounter (\n -> (n + 1, n)) + when (handlerCount == 25) $ + throwIO $ + DeliberateException $ SomeServerException 1 + incUntilFinal call + ] + } + +-- | This is essentially 'Test.Prop.Dialogue.earlyTermination15', but the server +-- does not wait for client termination. +test_earlyTerminationNoWait :: IO () +test_earlyTerminationNoWait = testClientServer $ ClientServerTest { + config = def + , client = simpleTestClient $ \conn -> do + _mResult <- + try @DeliberateException $ + Client.withRPC conn def (Proxy @Trivial) $ \_call -> + throwIO (DeliberateException $ SomeServerException 0) + + Client.withRPC conn def (Proxy @Trivial) $ \call -> do + Binary.sendFinalInput @Word8 call 0 + _output <- Binary.recvOutput @Word8 call + return () + , server = [ + Server.someRpcHandler $ + Server.mkRpcHandler @Trivial $ \call -> + Binary.recvInput @Word8 call >>= \case + _ -> Server.sendTrailers call NoMetadata + ] + } + +{------------------------------------------------------------------------------- + Client and handler functions +-------------------------------------------------------------------------------} + +-- | Send numbers to the server until it responds with one that satisfies the +-- given predicate. +countUntil :: (Word64 -> Bool) -> Client.Call Trivial -> IO Word64 +countUntil = go 0 + where + go :: Word64 -> (Word64 -> Bool) -> Client.Call Trivial -> IO Word64 + go next p call + | p next + = do + Binary.sendFinalInput @Word64 call next + (_final, NoMetadata) <- Binary.recvFinalOutput @Word64 call + return next + | otherwise + = do + Binary.sendNextInput @Word64 call next + next' <- Binary.recvNextOutput @Word64 call + go next' p call + +-- | Reads numbers from the client and sends them back incremented by one. +incUntilFinal :: Server.Call Trivial -> IO () +incUntilFinal call = do + Binary.recvInput call >>= \case + StreamElem n -> do + Binary.sendNextOutput @Word64 call $ succ n + incUntilFinal call + FinalElem n _ -> do + Binary.sendFinalOutput @Word64 call (succ n, NoMetadata) + NoMoreElems _ -> do + -- TODO: + -- + -- We shouldn't need to handle this case, since our client never + -- explicitly sends 'NoMoreElems'. However, see discussion in the + -- ticket above. + Server.sendTrailers call NoMetadata + return () + +{------------------------------------------------------------------------------- + Auxiliary +-------------------------------------------------------------------------------} + +type Trivial = RawRpc "trivial" "trivial" + +type instance RequestMetadata Trivial = NoMetadata +type instance ResponseInitialMetadata Trivial = NoMetadata +type instance ResponseTrailingMetadata Trivial = NoMetadata diff --git a/test-grapesy/Test/Util.hs b/test-grapesy/Test/Util.hs index f5510ba3..ef8e0b3e 100644 --- a/test-grapesy/Test/Util.hs +++ b/test-grapesy/Test/Util.hs @@ -4,6 +4,9 @@ module Test.Util ( -- * Timeouts Timeout(..) , within + + -- * Files + , withTemporaryFile ) where import Control.Concurrent @@ -11,6 +14,8 @@ import Control.Exception import Control.Monad.Catch import Control.Monad.IO.Class import GHC.Stack +import System.Directory +import System.IO {------------------------------------------------------------------------------- Timeouts @@ -45,4 +50,13 @@ within t info io = do fmap fst $ generalBracket startTimer stopTimer $ \_ -> io - +withTemporaryFile :: (FilePath -> IO a) -> IO a +withTemporaryFile k = do + tmpDir <- getTemporaryDirectory + Control.Exception.bracket + (openTempFile tmpDir "grapesy-test-suite.txt") + (removeFile . fst) + ( \(fp, h) -> do + hClose h + k fp + ) diff --git a/test-grapesy/Test/Util/Exception.hs b/test-grapesy/Test/Util/Exception.hs new file mode 100644 index 00000000..5b0af1f8 --- /dev/null +++ b/test-grapesy/Test/Util/Exception.hs @@ -0,0 +1,36 @@ +-- | Utility exception types for the tests +module Test.Util.Exception + ( -- * User exceptions + SomeServerException(..) + , SomeClientException(..) + + -- * Deliberate exceptions + , DeliberateException(..) + , ExceptionId + ) where + +import Control.Exception + +{------------------------------------------------------------------------------- + User exceptions + + When a test calls for the client or the server to throw an exception, we throw + one of these. Their sole purpose is to be "any" kind of exception (not a + specific one). +-------------------------------------------------------------------------------} + +data SomeServerException = SomeServerException ExceptionId + deriving stock (Show, Eq) + deriving anyclass (Exception) + +data SomeClientException = SomeClientException ExceptionId + deriving stock (Show, Eq) + deriving anyclass (Exception) + +-- | Exception thrown by client or handler to test exception handling +data DeliberateException = forall e. Exception e => DeliberateException e + deriving anyclass (Exception) +deriving stock instance Show DeliberateException + +-- | We distinguish exceptions from each other simply by a number +type ExceptionId = Int diff --git a/test-grapesy/Test/Util/RawTestServer.hs b/test-grapesy/Test/Util/RawTestServer.hs new file mode 100644 index 00000000..10d5708e --- /dev/null +++ b/test-grapesy/Test/Util/RawTestServer.hs @@ -0,0 +1,110 @@ +module Test.Util.RawTestServer where + +import Control.Concurrent +import Control.Concurrent.Async +import Control.Exception +import Data.ByteString qualified as BS.Strict +import Data.ByteString qualified as Strict (ByteString) +import Data.ByteString.Builder qualified as BS.Builder +import Data.ByteString.Char8 qualified as BS.Strict.Char8 +import Data.ByteString.UTF8 qualified as BS.Strict.UTF8 +import Data.String (fromString) +import Network.HTTP2.Server qualified as HTTP2 +import Network.Run.TCP qualified as NetworkRun +import Network.Socket + +import Network.GRPC.Client qualified as Client +import Network.HTTP.Types qualified as HTTP +import Network.GRPC.Common +import Data.Maybe + +{------------------------------------------------------------------------------- + Raw test server + + This allows us to simulate broken /servers/. +-------------------------------------------------------------------------------} + +-- | Low-level test server +-- +-- We bypass the entire grapesy machinery for constructing the server, for added +-- flexibility. This allows us to mock broken deployments or run the server in +-- another thread that we throw asynchronous exceptions to, for example. +-- +-- The grapesy client can auto reconnect when the server is not (yet) up and +-- running, but to keep things simple, we just signal when the server is ready. +-- This also allows us to avoid binding to a specific port in the tests (which +-- might already be in use on the machine running the tests, leading to spurious +-- test failures). +rawTestServer :: IO (Maybe PortNumber) -> (PortNumber -> IO ()) -> HTTP2.Server -> IO () +rawTestServer getPort signalPort server = do + mPortIn <- fromMaybe 0 <$> getPort + addr <- NetworkRun.resolve Stream (Just "127.0.0.1") (show mPortIn) [AI_PASSIVE] + bracket (NetworkRun.openTCPServerSocket addr) close $ \listenSock -> do + addr' <- getSocketName listenSock + portOut <- case addr' of + SockAddrInet port _host -> return port + SockAddrInet6 port _ _host _ -> return port + SockAddrUnix{} -> error "rawTestServer: unexpected unix socket" + signalPort portOut + NetworkRun.runTCPServerWithSocket listenSock $ \clientSock -> + bracket (HTTP2.allocSimpleConfig clientSock 4096) + HTTP2.freeSimpleConfig $ \config -> + HTTP2.run HTTP2.defaultServerConfig config server + +-- | Run the server and apply the continuation to an 'Client.Address' holding +-- the running server's host and port. +withTestServer :: HTTP2.Server -> (Client.Address -> IO a) -> IO a +withTestServer server k = do + serverPort <- newEmptyMVar + withAsync (rawTestServer (pure Nothing) (putMVar serverPort) server) $ + \_serverThread -> do + port <- readMVar serverPort + let addr :: Client.Address + addr = Client.Address { + addressHost = "127.0.0.1" + , addressPort = port + , addressAuthority = Nothing + } + k addr + +-- | Server that responds with the given 'Response', independent of the request +respondWith :: Response -> (Client.Address -> IO a) -> IO a +respondWith response = withTestServer $ \_req _aux respond -> + respond (toHTTP2Response response) [] + +data Response = Response { + responseStatus :: HTTP.Status + , responseHeaders :: [HTTP.Header] + , responseBody :: Strict.ByteString + , responseTrailers :: [HTTP.Header] + } + +instance Default Response where + def = Response { + responseStatus = HTTP.ok200 + , responseHeaders = [ asciiHeader "content-type" "application/grpc" ] + , responseBody = BS.Strict.empty + , responseTrailers = [ asciiHeader "grpc-status" "0" ] + } + +toHTTP2Response :: Response -> HTTP2.Response +toHTTP2Response response = + flip HTTP2.setResponseTrailersMaker trailersMaker $ + HTTP2.responseBuilder + (responseStatus response) + (responseHeaders response) + (BS.Builder.byteString $ responseBody response) + where + trailersMaker :: HTTP2.TrailersMaker + trailersMaker Nothing = return $ HTTP2.Trailers (responseTrailers response) + trailersMaker (Just _) = return $ HTTP2.NextTrailersMaker trailersMaker + +-- | Header with ASCII value +-- +-- (Header /names/ are always ASCII.) +asciiHeader :: String -> String -> HTTP.Header +asciiHeader name value = (fromString name, BS.Strict.Char8.pack value) + +-- | Header with UTF-8 encoded value +utf8Header :: String -> String -> HTTP.Header +utf8Header name value = (fromString name, BS.Strict.UTF8.fromString value) diff --git a/util/Network/GRPC/Util/Session/Channel.hs b/util/Network/GRPC/Util/Session/Channel.hs index 56bd1eff..ebb87de9 100644 --- a/util/Network/GRPC/Util/Session/Channel.hs +++ b/util/Network/GRPC/Util/Session/Channel.hs @@ -429,16 +429,16 @@ close Channel{channelOutbound} reason = do -- We leave the inbound thread running. Although the channel is closed, -- there might still be unprocessed messages in the queue. The inbound -- thread will terminate once it reaches the end of the queue - outbound <- cancelThread channelOutbound channelClosed - case outbound of - AlreadyTerminated _ -> - return $ Nothing - AlreadyAborted _err -> - -- Connection_ to the peer was lost prior to closing - return $ Nothing - Cancelled -> - -- Proper procedure for outbound messages was not followed - return $ Just channelClosed + outbound <- cancelThread channelOutbound channelClosed + case outbound of + AlreadyTerminated _ -> + return $ Nothing + AlreadyAborted _err -> + -- Connection_ to the peer was lost prior to closing + return $ Nothing + Cancelled -> + -- Proper procedure for outbound messages was not followed + return $ Just channelClosed where channelClosed :: SomeException channelClosed = @@ -522,7 +522,7 @@ linkOutboundToInbound allowHalfClosed channel inbound = do threads, using the 'Thread' API from "Network.GRPC.Util.Thread". We are therefore not particularly worried about these loops being interrupted by asynchronous exceptions: this only happens if the threads are explicitly - terminated (when the corrresponding channels are closed), in which case any + terminated (when the corresponding channels are closed), in which case any attempt to interact with them after they have been killed will be handled by 'getThreadInterface' throwing 'ThreadInterfaceUnavailable'. -------------------------------------------------------------------------------}