quic-0.3.14: test/IOSpec.hs

{-# LANGUAGE NumericUnderscores #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE ScopedTypeVariables #-}

module IOSpec where

import Control.Concurrent
import Control.Concurrent.Async
import qualified Control.Exception as E
import Control.Monad
import qualified Data.ByteString as BS
import Data.IORef
import qualified System.Timeout as Timeout
import Test.Hspec

import Network.QUIC
import qualified Network.QUIC.Client as C
import Network.QUIC.Internal
import Network.QUIC.Server

import Config

spec :: Spec
spec = do
    runIO prepareQlog
    sc0 <- runIO makeTestServerConfigR
    var <- runIO newEmptyMVar
    let sc =
            sc0
                { scHooks =
                    (scHooks sc0)
                        { onServerReady = putMVar var ()
                        }
                }
    let cc = setClientQlog testClientConfigR
    -- With a timeout: 'run' reports a server it could not start, but it is
    -- reported to whoever forked it, and that is not us.  Without this the
    -- take never returns and the whole suite stops rather than failing.
    let waitS =
            Timeout.timeout 5000000 (takeMVar var) >>= \r -> case r of
                Just () -> return ()
                Nothing -> expectationFailure "server never became ready"
    describe "handshake" $ do
        it "can exchange data when the server's first flight is lost" $ do
            withPipe (DropServerPacket [0, 1, 2]) $ testSendRecv cc sc waitS 20
    describe "send & recv" $ do
        it "can exchange data on random dropping" $ do
            withPipe (Randomly 20) $ testSendRecv cc sc waitS 1000
        it "can exchange data on server 0" $ do
            withPipe (DropServerPacket [0]) $ testSendRecv cc sc waitS 20
        it "can exchange data on server 1" $ do
            withPipe (DropServerPacket [1]) $ testSendRecv cc sc waitS 20
        it "can exchange data on server 2" $ do
            withPipe (DropServerPacket [2]) $ testSendRecv cc sc waitS 20
        it "can exchange data on server 3" $ do
            withPipe (DropServerPacket [3]) $ testSendRecv cc sc waitS 20
        it "can exchange data on server 4" $ do
            withPipe (DropServerPacket [4]) $ testSendRecv cc sc waitS 20
        it "can exchange data on server 5" $ do
            withPipe (DropServerPacket [5]) $ testSendRecv cc sc waitS 20
        it "can exchange data on server 6" $ do
            withPipe (DropServerPacket [6]) $ testSendRecv cc sc waitS 20
        it "can exchange data on server 7" $ do
            withPipe (DropServerPacket [7]) $ testSendRecv cc sc waitS 20
        it "can exchange data on server 8" $ do
            withPipe (DropServerPacket [8]) $ testSendRecv cc sc waitS 20
        it "can exchange data on server 9" $ do
            withPipe (DropServerPacket [9]) $ testSendRecv cc sc waitS 20
        it "can exchange data on server 10" $ do
            withPipe (DropServerPacket [10]) $ testSendRecv cc sc waitS 20
        it "can exchange data on server 11" $ do
            withPipe (DropServerPacket [11]) $ testSendRecv cc sc waitS 20
        it "can exchange data on client 0" $ do
            withPipe (DropClientPacket [0]) $ testSendRecv cc sc waitS 20
        it "can exchange data on client 1" $ do
            withPipe (DropClientPacket [1]) $ testSendRecv cc sc waitS 20
        it "can exchange data on client 2" $ do
            withPipe (DropClientPacket [2]) $ testSendRecv cc sc waitS 20
        it "can exchange data on client 3" $ do
            withPipe (DropClientPacket [3]) $ testSendRecv cc sc waitS 20
        it "can exchange data on client 4" $ do
            withPipe (DropClientPacket [4]) $ testSendRecv cc sc waitS 20
        it "can exchange data on client 5" $ do
            withPipe (DropClientPacket [5]) $ testSendRecv cc sc waitS 20
        it "can exchange data on client 6" $ do
            withPipe (DropClientPacket [6]) $ testSendRecv cc sc waitS 20
        it "can exchange data on client 7" $ do
            withPipe (DropClientPacket [7]) $ testSendRecv cc sc waitS 20
        it "can exchange data on client 8" $ do
            withPipe (DropClientPacket [8]) $ testSendRecv cc sc waitS 20
        it "can exchange data on client 9" $ do
            withPipe (DropClientPacket [9]) $ testSendRecv cc sc waitS 20
        it "can exchange data on client 10" $ do
            withPipe (DropClientPacket [10]) $ testSendRecv cc sc waitS 20
        it "can exchange data on client 11" $ do
            withPipe (DropClientPacket [11]) $ testSendRecv cc sc waitS 20
    describe "recvStream" $ do
        -- https://github.com/kazu-yamamoto/quic/pull/54
        it "don't block if client stop sending first" $ do
            withPipe (Randomly 20) $ testRecvStreamClientStopFirst cc sc waitS
        it "don't block if server stop sending first" $ do
            withPipe (Randomly 20) $ testRecvStreamServerStopFirst cc sc waitS
        -- RFC 9000: https://www.rfc-editor.org/rfc/rfc9000.html
        -- Section 3.5 says STOP_SENDING asks the peer to send RESET_STREAM.
        -- Sections 4.5 and 19.4 define RESET_STREAM Final Size as the
        -- number of bytes sent by the RESET_STREAM sender.
        it "sends RESET_STREAM with the bytes sent as final size" $ do
            withPipe (DropClientPacket []) $ testResetStreamFinalSize cc sc waitS
        it "tells a stream that was reset from one that ended" $ do
            withPipe (DropClientPacket []) $ testResetReceived cc sc waitS
    describe "closed stream" $ do
        it "ignores a late copy of the data it received on a stream it opened" $ do
            withPipe (DelayServerPacket 300) $ testLateCopy False cc sc waitS
        it "ignores a late copy of the data it received on a stream the peer opened" $ do
            withPipe (DelayClientPacket 300) $ testLateCopy True cc sc waitS
    describe "stream limits" $ do
        it "keeps the peer's open bidirectional streams within the limit" $ do
            withPipe (DropClientPacket []) $ testOpenStreams cc sc waitS
        it "raises the limit on unidirectional streams by their own number" $ do
            withPipe (DropClientPacket []) $ testUniStreams cc sc waitS
    describe "stream state" $ do
        it "can still send after the peer's RESET_STREAM" $ do
            withPipe (DropClientPacket []) $ testSendAfterReset cc sc waitS
        it "cannot send after the peer's STOP_SENDING" $ do
            withPipe (DropClientPacket []) $ testStopSending cc sc waitS
        it "ends the stream for the reader when it is reset" $ do
            withPipe (DropClientPacket []) $ testRecvAfterReset cc sc waitS
        it "accepts a stream the peer only reset" $ do
            withPipe (DropClientPacket []) $ testResetOnly cc sc waitS
        it "accepts a reset that overtook the data it followed" $ do
            withPipe (DropClientPacket []) $ testResetOvertakes cc sc waitS
        it "accepts a stream the peer is only blocked on" $ do
            withPipe (DropClientPacket []) $ testDataBlockedOpens cc sc waitS
    describe "final size" $ do
        it "refuses a final size that is not the one already known" $ do
            withPipe (DropClientPacket []) $
                testFinalSize cc sc waitS [StreamF 0 0 ["ab"] True, StreamF 0 0 ["abcd"] True]
        it "refuses a final size below what the stream has reached" $ do
            withPipe (DropClientPacket []) $
                testFinalSize cc sc waitS [StreamF 0 0 ["abcd"] False, StreamF 0 0 ["ab"] True]
        it "refuses stream data beyond the final size" $ do
            withPipe (DropClientPacket []) $
                testFinalSize cc sc waitS [StreamF 0 0 ["ab"] True, StreamF 0 2 ["cd"] False]
        it "counts what the application never reads against the connection" $ do
            withPipe (DropClientPacket []) $ testCloseWithoutReading cc sc waitS
        it "counts a reset stream's final size against the connection" $ do
            withPipe (DropClientPacket []) $ testResetCountsAgainstTheWindow cc sc waitS
    describe "closing" $ do
        it "tells the peer when the application gives up" $ do
            withPipe (DropClientPacket []) $ testServerThrows cc sc waitS
        it "tells the peer before waiting for the application" $ do
            withPipe (DropClientPacket []) $ testSlowApp cc sc waitS
    describe "port handover" $ do
        it "ignores a leftover datagram from the connection that just closed" $
            withPipeStray (Randomly 20) $
                testSendRecv cc sc waitS 20
    describe "concurrency" $ do
        it "can handle multiple clients" $ do
            withPipe (Randomly 20) $ testMultiSendRecv cc sc waitS 500
    describe "packing" $ do
        it "fits the frames of many streams with tiny writes into a packet" $ do
            withPipe (DropClientPacket []) $ testTinyWrites cc sc waitS
    describe "abortConnection" $ do
        it "can abort connection" $ do
            withPipe (Randomly 20) $ testAbort cc sc waitS

-- | Writing data to many, many streams doesn't cause a buffer overrun
testTinyWrites :: C.ClientConfig -> ServerConfig -> IO () -> IO ()
testTinyWrites cc sc0 waitS = do
    doneVar <- newChan
    withAsync (server doneVar) $ \_ -> client doneVar
  where
    nStreams = 400
    sc =
        sc0
            { scParameters =
                (scParameters sc0){initialMaxStreamsBidi = 2 * nStreams}
            }

    client doneVar = do
        waitS
        C.run cc $ \conn -> do
            strms <- replicateM nStreams (stream conn)
            forM_ strms $ \strm -> sendStream strm "a"
            threadDelay 200_000
            forM_ strms $ \strm -> sendStream strm "b"
            -- Wait for the server to have read both bytes of every stream
            -- before closing the connection.
            mres <- Timeout.timeout 10_000_000 $ replicateM_ nStreams $ readChan doneVar
            mres `shouldBe` Just ()

    server doneVar = run sc $ \conn -> forever $ do
        strm <- acceptStream conn
        void $ forkIO $ do
            consumeBytes strm 2
            writeChan doneVar ()

consumeBytes :: Stream -> Int -> IO ()
consumeBytes _ 0 = return ()
consumeBytes strm left = do
    bs <- recvStream strm 1024
    when (BS.null bs) $ expectationFailure "no enough bytes received"
    let len = BS.length bs
    when (len > left) $ expectationFailure "extra bytes received"
    consumeBytes strm (left - len)

assertEndOfStream :: Stream -> IO ()
assertEndOfStream strm = recvStream strm 1024 `shouldReturn` ""

testResetStreamFinalSize
    :: C.ClientConfig -> ServerConfig -> IO () -> IO ()
testResetStreamFinalSize cc0 sc waitS = do
    finalSizeVar <- newEmptyMVar
    doneVar <- newEmptyMVar
    let request = "open"
        payload = BS.replicate 1234 0
        hooks = (ccHooks cc0){onResetStreamReceived2 = record finalSizeVar}
        cc = cc0{ccHooks = hooks}
    withAsync (server request payload doneVar) $ \_ ->
        client cc request payload finalSizeVar doneVar
  where
    aerr = ApplicationProtocolError 0

    record finalSizeVar _strm _aerr finalSize = void $ tryPutMVar finalSizeVar finalSize

    client cc request payload finalSizeVar doneVar = do
        waitS
        C.run cc $ \conn -> do
            strm <- stream conn
            sendStream strm request
            consumeBytes strm (BS.length payload)
            stopStream strm aerr
            mres <- Timeout.timeout 5000000 $ takeMVar finalSizeVar
            mres `shouldBe` Just (BS.length payload)
            putMVar doneVar ()

    server request payload doneVar = run sc $ \conn -> do
        strm <- acceptStream conn
        consumeBytes strm (BS.length request)
        sendStream strm payload
        takeMVar doneVar

-- | One end has read a stream to its end and closed it while a packet for
--   it is still on the way.  That packet was taken for lost and its data sent
--   again, so what arrives late is a copy, well past the initial window of
--   256K.  It must not open the stream anew: the new stream holds the copy to
--   that window and calls it a flow control error, or else hands the
--   application a stream that was never opened.
testLateCopy :: Bool -> C.ClientConfig -> ServerConfig -> IO () -> IO ()
testLateCopy upload cc sc waitS =
    withAsync server $ \_ -> client
  where
    (upLen, downLen)
        | upload = (1000000, 10)
        | otherwise = (10, 1000000)

    client = do
        waitS
        C.run cc $ \conn -> do
            exchange conn
            -- The copy arrives while we wait.
            threadDelay 200000
            exchange conn
    exchange conn = do
        strm <- stream conn
        sendStream strm (BS.replicate upLen 0)
        shutdownStream strm
        consumeBytes strm downLen
        assertEndOfStream strm
        closeStream strm

    server = run sc $ \conn -> forever $ do
        strm <- acceptStream conn
        consumeBytes strm upLen
        assertEndOfStream strm
        sendStream strm (BS.replicate downLen 0)
        closeStream strm

-- | After a RESET_STREAM, 'recvStream' returns "" just as at the end of the
-- stream; 'resetReceived' tells the two apart.  The client ends one stream
-- and resets another, and the server looks at both.
testResetReceived :: C.ClientConfig -> ServerConfig -> IO () -> IO ()
testResetReceived cc sc waitS = do
    resultVar <- newEmptyMVar
    withAsync (server resultVar) $ \_ -> client resultVar
  where
    aerr = ApplicationProtocolError 7

    client resultVar = do
        waitS
        C.run cc $ \conn -> do
            strm0 <- stream conn
            sendStream strm0 "ended"
            shutdownStream strm0
            strm1 <- stream conn
            sendStream strm1 "reset"
            -- Let the data go first.
            threadDelay 100000
            resetStream strm1 aerr
            mres <- Timeout.timeout 5000000 $ takeMVar resultVar
            mres `shouldBe` Just (Nothing, Just aerr)

    server resultVar = run sc $ \conn -> do
        strm0 <- acceptStream conn
        strm1 <- acceptStream conn
        let drain strm = do
                bs <- recvStream strm 1024
                unless (BS.null bs) $ drain strm
        drain strm0
        drain strm1
        r0 <- resetReceived strm0
        r1 <- resetReceived strm1
        putMVar resultVar (r0, r1)
        threadDelay 1000000

-- | The server closes one stream in ten and keeps the others open; the
-- client opens streams for as long as it is let.  No more than
-- initial_max_streams_bidi may be open at once (RFC 9000, section 4.6).
--
-- The limit used to be raised to the highest stream opened plus the initial
-- number whenever a stream was closed, so that the client was given a whole
-- new window each time: over three thousand were held open within seconds.
testOpenStreams :: C.ClientConfig -> ServerConfig -> IO () -> IO ()
testOpenStreams cc sc waitS = do
    maxOpen <- newIORef (0 :: Int)
    withAsync (server maxOpen) $ \_ -> do
        client
        readIORef maxOpen
            >>= (`shouldSatisfy` (<= initialMaxStreamsBidi (scParameters sc)))
  where
    client = do
        waitS
        C.run cc $ \conn -> do
            _ <- Timeout.timeout 1500000 $ forever $ do
                strm <- stream conn
                sendStream strm "x"
            return ()
    server maxOpen = run sc $ \conn -> do
        openRef <- newIORef (0 :: Int)
        forM_ [0 :: Int ..] $ \n -> do
            strm <- acceptStream conn
            _ <- recvStream strm 1
            if n `mod` 10 == 0
                then closeStream strm
                else do
                    o <- atomicModifyIORef' openRef $ \x -> (x + 1, x + 1)
                    atomicModifyIORef' maxOpen $ \m -> (max m o, ())

-- | The server closes the first unidirectional stream and keeps the rest.
-- The client may then have initial_max_streams_uni plus one.  The limit was
-- raised by initial_max_streams_bidi instead, 64 against 3.
testUniStreams :: C.ClientConfig -> ServerConfig -> IO () -> IO ()
testUniStreams cc sc waitS = do
    opened <- newIORef (0 :: Int)
    withAsync server $ \_ -> do
        client opened
        readIORef opened
            >>= (`shouldSatisfy` (<= initialMaxStreamsUni (scParameters sc) + 1))
  where
    client opened = do
        waitS
        C.run cc $ \conn -> do
            _ <- Timeout.timeout 1500000 $ forever $ do
                strm <- unidirectionalStream conn
                sendStream strm "x"
                modifyIORef' opened (+ 1)
            return ()
    server = run sc $ \conn -> do
        strm0 <- acceptStream conn
        _ <- recvStream strm0 1
        closeStream strm0
        forever $ do
            strm <- acceptStream conn
            recvStream strm 1

testRecvStreamClientStopFirst
    :: C.ClientConfig -> ServerConfig -> IO () -> IO ()
testRecvStreamClientStopFirst cc sc waitS = do
    mvar <- newEmptyMVar
    withAsync (server mvar) $ \_ -> client mvar
  where
    aerr = ApplicationProtocolError 0

    client mvar = do
        waitS
        C.run cc $ \conn -> do
            strm <- stream conn
            sendStream strm (BS.replicate 10000 0)
            takeMVar mvar `shouldReturn` ()
            stopStream strm aerr
            resetStream strm aerr
            takeMVar mvar `shouldReturn` ()
    server mvar = run sc $ \conn -> do
        strm <- acceptStream conn
        consumeBytes strm 10000 `shouldReturn` ()
        -- notify client to stop stream after all bytes are received.
        putMVar mvar ()
        -- verify that recvStream does not block
        assertEndOfStream strm
        putMVar mvar ()

testRecvStreamServerStopFirst
    :: C.ClientConfig -> ServerConfig -> IO () -> IO ()
testRecvStreamServerStopFirst cc sc waitS = do
    mvar <- newEmptyMVar
    withAsync (server mvar) $ \_ -> client mvar
  where
    aerr = ApplicationProtocolError 0

    client mvar = do
        waitS
        C.run cc $ \conn -> do
            strm <- stream conn
            sendStream strm (BS.replicate 10000 0)
            takeMVar mvar `shouldReturn` ()
    server mvar = run sc $ \conn -> do
        strm <- acceptStream conn
        consumeBytes strm 10000 `shouldReturn` ()
        -- ask client to stop sending.
        stopStream strm aerr
        -- verify that recvStream does not block
        assertEndOfStream strm
        resetStream strm aerr
        putMVar mvar ()

testSendRecv :: C.ClientConfig -> ServerConfig -> IO () -> Int -> IO ()
testSendRecv cc sc waitS times = do
    mvar <- newEmptyMVar
    withAsync (server mvar) $ \_ -> client mvar
  where
    client mvar = do
        waitS
        C.run cc $ \conn -> do
            strm <- stream conn
            let bs = BS.replicate 10000 0
            replicateM_ times $ sendStream strm bs
            shutdownStream strm
            takeMVar mvar `shouldReturn` ()
    server mvar = run sc $ \conn -> do
        strm <- acceptStream conn
        consumeBytes strm (10000 * times)
        assertEndOfStream strm
        putMVar mvar ()

testMultiSendRecv :: C.ClientConfig -> ServerConfig -> IO () -> Int -> IO ()
testMultiSendRecv cc sc waitS times = do
    mvars <- replicateM concurrency newEmptyMVar
    withAsync (server mvars) $ \_ -> client mvars
  where
    concurrency = 10
    chunklen = 12345
    chunk = BS.replicate chunklen 0
    client mvars = do
        waitS
        C.run cc $ \conn -> foldr1 concurrently_ $ replicate concurrency $ go conn
      where
        go conn = do
            strm <- stream conn
            let n = streamId strm `div` 4
            sendStream strm $ BS.singleton $ fromIntegral n
            replicateM_ times $ sendStream strm chunk
            shutdownStream strm
            takeMVar (mvars !! n) `shouldReturn` ()
    server mvars = run sc loop
      where
        loop conn = do
            strm <- acceptStream conn
            void $ forkIO $ do
                bs <- recvStream strm 1
                case BS.uncons bs of
                    Nothing -> return ()
                    Just (w, _) -> do
                        let n = fromIntegral w
                        consumeBytes strm (chunklen * times)
                        assertEndOfStream strm
                        putMVar (mvars !! n) ()
            loop conn

appErr :: QUICException -> Bool
appErr (ApplicationProtocolErrorIsReceived _ _) = True
appErr _ = False

testAbort :: C.ClientConfig -> ServerConfig -> IO () -> IO ()
testAbort cc sc waitS = do
    withAsync server $ \_ ->
        client `shouldThrow` appErr
  where
    client = do
        waitS
        C.run cc $ \conn -> do
            strm <- stream conn
            sendStream strm "foo"
            void $ recvStream strm 10
    server = run sc loop
      where
        loop conn = do
            _strm <- acceptStream conn
            void $
                forkIO $
                    abortConnection
                        conn
                        (ApplicationProtocolError 1)
                        "testing abortConnection"
            loop conn

-- | RESET_STREAM ends the peer's sending part alone (RFC 9000, section 3.2).
--   What we have to send on a bidirectional stream must still go out.
testSendAfterReset :: C.ClientConfig -> ServerConfig -> IO () -> IO ()
testSendAfterReset cc sc waitS = do
    res <- newEmptyMVar
    withAsync (server res) $ \_ -> do
        client
        takeMVar res >>= (`shouldBe` True)
  where
    client = do
        waitS
        C.run cc $ \conn -> do
            strm <- stream conn
            sendStream strm "ping"
            -- The reply says the "ping" has arrived, so that the
            -- RESET_STREAM cannot overtake it and be dropped as a frame
            -- for a stream that does not exist yet.
            _ <- recvStream strm 1024
            resetStream strm (ApplicationProtocolError 0)
            threadDelay 300000
    server res = run sc $ \conn -> do
        strm <- acceptStream conn
        _ <- recvStream strm 1024
        sendStream strm "ack"
        -- An empty ByteString: the pseudo FIN that the RESET_STREAM left.
        let recvEOF = do
                bs <- recvStream strm 1024
                unless (bs == "") recvEOF
        recvEOF
        sent <-
            (True <$ sendStream strm "pong") `E.catch` \e -> case e of
                StreamIsClosed -> return False
                _ -> E.throwIO e
        putMVar res sent

-- | STOP_SENDING makes us reset our sending part (RFC 9000, section 3.5),
--   so nothing more may go out on it.
testStopSending :: C.ClientConfig -> ServerConfig -> IO () -> IO ()
testStopSending cc sc waitS =
    withAsync server $ \_ -> client
  where
    client = do
        waitS
        C.run cc $ \conn -> do
            strm <- stream conn
            sendStream strm "ping"
            stopped <-
                ( do
                    _ <- Timeout.timeout 2000000 $ forever $ do
                        threadDelay 20000
                        sendStream strm "x"
                    return False
                )
                    `E.catch` \e -> case e of
                        StreamIsClosed -> return True
                        _ -> E.throwIO e
            stopped `shouldBe` True
    server = run sc $ \conn -> do
        strm <- acceptStream conn
        _ <- recvStream strm 1024
        stopStream strm (ApplicationProtocolError 0)
        threadDelay 3000000

-- | 'resetStream' leaves nothing to wait for.  An application that resets a
--   stream because 'sendStream' failed does so with the sending part closed
--   by the peer's STOP_SENDING already, and 'recvStream' has to end there
--   rather than block: what the peer sends afterwards is dropped anyway,
--   the stream being out of the table.
testRecvAfterReset :: C.ClientConfig -> ServerConfig -> IO () -> IO ()
testRecvAfterReset cc sc waitS =
    withAsync server $ \_ -> client
  where
    client = do
        waitS
        C.run cc $ \conn -> do
            strm <- stream conn
            sendStream strm "ping"
            _ <-
                ( do
                    _ <- Timeout.timeout 2000000 $ forever $ do
                        threadDelay 20000
                        sendStream strm "x"
                    return ()
                )
                    `E.catch` \e -> case e of
                        StreamIsClosed -> return ()
                        _ -> E.throwIO e
            resetStream strm (ApplicationProtocolError 0)
            mbs <- Timeout.timeout 1000000 $ recvStream strm 1024
            mbs `shouldBe` Just ""
    server = run sc $ \conn -> do
        strm <- acceptStream conn
        _ <- recvStream strm 1024
        stopStream strm (ApplicationProtocolError 0)
        threadDelay 3000000

-- | RFC 9000, section 3.2: the receiving part of a stream the peer opened is
--   created when the first STREAM, STREAM_DATA_BLOCKED or RESET_STREAM frame
--   for it arrives.  Here the client never sends on the stream, so the
--   RESET_STREAM is the first the server hears of it.
testResetOnly :: C.ClientConfig -> ServerConfig -> IO () -> IO ()
testResetOnly cc sc waitS = do
    got <- newEmptyMVar
    withAsync (server got) $ \_ -> do
        client
        -- With a timeout: without the stream, 'acceptStream' on the server
        -- never returns, and the test would hang rather than fail.
        r <- Timeout.timeout 1000000 $ takeMVar got
        r `shouldBe` Just (Just aerr)
  where
    aerr = ApplicationProtocolError 7
    client = do
        waitS
        C.run cc $ \conn -> do
            strm <- stream conn
            resetStream strm aerr
            threadDelay 300000
    server got = run sc $ \conn -> do
        strm <- acceptStream conn
        bs <- recvStream strm 1024
        bs `shouldBe` ""
        resetReceived strm >>= putMVar got

-- | The same, for the reset that follows data on the stream.  It overtakes
--   the data: the sender empties the queue the RESET_STREAM is on before the
--   one the STREAM frame is on.
testResetOvertakes :: C.ClientConfig -> ServerConfig -> IO () -> IO ()
testResetOvertakes cc sc waitS = do
    got <- newEmptyMVar
    withAsync (server got) $ \_ -> do
        client
        -- With a timeout: without the stream, 'acceptStream' on the server
        -- never returns, and the test would hang rather than fail.
        r <- Timeout.timeout 1000000 $ takeMVar got
        r `shouldBe` Just (Just aerr)
  where
    aerr = ApplicationProtocolError 7
    client = do
        waitS
        C.run cc $ \conn -> do
            strm <- stream conn
            sendStream strm "ping"
            resetStream strm aerr
            threadDelay 300000
    server got = run sc $ \conn -> do
        strm <- acceptStream conn
        let recvEOF = do
                bs <- recvStream strm 1024
                unless (bs == "") recvEOF
        recvEOF
        resetReceived strm >>= putMVar got

-- | RFC 9000, section 3.2 again, for the other frame of the three: a
--   STREAM_DATA_BLOCKED opens the stream it names as well.
testDataBlockedOpens :: C.ClientConfig -> ServerConfig -> IO () -> IO ()
testDataBlockedOpens cc0 sc waitS = do
    got <- newEmptyMVar
    withAsync (server got) $ \_ -> do
        client
        r <- Timeout.timeout 1000000 $ takeMVar got
        r `shouldBe` Just 0
  where
    cc = cc0{ccHooks = (ccHooks cc0){onPlainCreated = blockedOnStream0}}
    blockedOnStream0 lvl plain
        | lvl == RTT1Level =
            plain{plainFrames = StreamDataBlocked 0 0 : plainFrames plain}
        | otherwise = plain
    client = do
        waitS
        -- No stream of its own: the frame above is all the server hears of
        -- stream 0.
        C.run cc $ \_conn -> threadDelay 300000
    server got = run sc $ \conn -> do
        strm <- acceptStream conn
        putMVar got $ streamId strm

-- | RFC 9000, section 4.5: where a stream ends, once said, cannot be said
--   differently, and nothing may arrive past it.  The frames go in by a hook,
--   since a well-behaved client sends none of these.
testFinalSize
    :: C.ClientConfig -> ServerConfig -> IO () -> [Frame] -> IO ()
testFinalSize cc0 sc waitS frames =
    withAsync quietServer $ \_ -> client `shouldThrow` finalSizeError
  where
    quietServer = run sc $ \conn -> forever $ void $ acceptStream conn
    cc = cc0{ccHooks = (ccHooks cc0){onPlainCreated = inject}}
    inject lvl plain
        | lvl == RTT1Level = plain{plainFrames = frames ++ plainFrames plain}
        | otherwise = plain
    client = do
        waitS
        C.run cc $ \_conn -> threadDelay 1000000

finalSizeError :: QUICException -> Bool
finalSizeError (TransportErrorIsReceived FinalSizeError _) = True
finalSizeError _ = False

flowControlError :: QUICException -> Bool
flowControlError (TransportErrorIsReceived FlowControlError _) = True
flowControlError _ = False

-- | RFC 9000, section 4.5: "A receiver SHOULD use the final size to account
--   for all bytes sent on the stream in its connection-level flow
--   controller."  A hundred thousand octets is well past what this server
--   allows for the whole connection, so a stream reset at that final size,
--   none of which arrives, is over the limit and should be answered as one.
--
-- Uncounted, as it was, it cost the peer nothing at all: the window we
-- advertise fell behind what the peer believed it had spent, by the tail of
-- every stream it reset, and nothing here noticed.
testResetCountsAgainstTheWindow
    :: C.ClientConfig -> ServerConfig -> IO () -> IO ()
testResetCountsAgainstTheWindow cc0 sc0 waitS =
    withAsync quietServer $ \_ -> client `shouldThrow` flowControlError
  where
    sc = sc0{scParameters = (scParameters sc0){initialMaxData = 1000}}
    quietServer = run sc $ \conn -> forever $ void $ acceptStream conn
    cc = cc0{ccHooks = (ccHooks cc0){onPlainCreated = inject}}
    inject lvl plain
        | lvl == RTT1Level =
            plain
                { plainFrames =
                    ResetStream 0 (ApplicationProtocolError 0) 100000 : plainFrames plain
                }
        | otherwise = plain
    client = do
        waitS
        C.run cc $ \_conn -> threadDelay 1000000

-- | RFC 9000, section 10.2: an endpoint that ends a connection sends
--   CONNECTION_CLOSE.  Anything thrown in here that is not already on its way
--   to the peer used to end the connection in silence, leaving the peer to
--   wait out its own idle timeout.
testServerThrows :: C.ClientConfig -> ServerConfig -> IO () -> IO ()
testServerThrows cc sc waitS =
    withAsync server $ \_ -> client `shouldThrow` internalError
  where
    server = run sc $ \_conn -> E.throwIO $ userError "the application gave up"
    client = do
        waitS
        C.run cc $ \conn -> do
            strm <- stream conn
            sendStream strm "ping"
            void $ recvStream strm 1024

internalError :: QUICException -> Bool
internalError (TransportErrorIsReceived InternalError _) = True
internalError _ = False

-- | The peer hears of a transport error at once, however long the
--   application takes to unwind.
--
-- The protocol threads and the application run under one 'concurrently_',
-- which cancels the application when they fail and then waits for it, under
-- 'uninterruptibleMask_'.  Said after that, the CONNECTION_CLOSE waits with
-- it, and a peer told nothing waits out its own idle timeout.  Here the
-- application sits for three seconds after it is cancelled and the client
-- allows one and a half.
testSlowApp :: C.ClientConfig -> ServerConfig -> IO () -> IO ()
testSlowApp cc0 sc waitS =
    withAsync server $ \_ -> client `shouldThrow` frameEncodingError
  where
    server = run sc $ \conn ->
        forever (void $ acceptStream conn) `E.catch` \(e :: E.SomeException) -> do
            threadDelay 3000000
            E.throwIO e
    cc = cc0{ccHooks = (ccHooks cc0){onPlainCreated = inject}}
    inject lvl plain
        | lvl == RTT1Level =
            plain
                { plainFrames = MaxStreams Bidirectional (2 ^ (61 :: Int)) : plainFrames plain
                }
        | otherwise = plain
    client = do
        waitS
        C.run cc $ \_conn -> threadDelay 1500000

frameEncodingError :: QUICException -> Bool
frameEncodingError (TransportErrorIsReceived FrameEncodingError _) = True
frameEncodingError _ = False

-- | A server that answers without reading the body still gives the octets
--   back to the connection's window.
--
-- They were counted as received and, read by nobody, were never counted as
-- consumed, so the window we advertise stayed that much smaller for the rest
-- of the connection.  Here twenty streams of four kilobytes go to a server
-- that reads one octet of each, against a window of thirty-two: uncounted,
-- the client is blocked before it is halfway through.
testCloseWithoutReading :: C.ClientConfig -> ServerConfig -> IO () -> IO ()
testCloseWithoutReading cc sc0 waitS =
    withAsync server $ \_ -> client
  where
    sc = sc0{scParameters = (scParameters sc0){initialMaxData = 32768}}
    server = run sc $ \conn -> forever $ do
        strm <- acceptStream conn
        _ <- recvStream strm 1
        closeStream strm
    client = do
        waitS
        r <- Timeout.timeout 5000000 $ C.run cc $ \conn ->
            replicateM_ 20 $ do
                strm <- stream conn
                sendStream strm $ BS.replicate 4096 0
                shutdownStream strm
                closeStream strm
        r `shouldBe` Just ()