Slow or stuck stream consumer can cause memory to blow up

[akshaymankar]

Here is an example with a stuck stream consumer:

{-# LANGUAGE OverloadedStrings #-}

module Main where

import Control.Concurrent
import Control.Concurrent.Async
import Control.Exception
import Control.Monad
import qualified Data.ByteString as BS
import qualified Data.ByteString.Builder as Builder
import qualified Data.ByteString.Lazy as LBS
import Data.Streaming.Network
import Network.HTTP.Types
import qualified Network.HTTP2.Client as Client
import qualified Network.HTTP2.Server as Server
import Network.Socket

testServer :: Server.Request -> Server.Aux -> (Server.Response -> [Server.PushPromise] -> IO ()) -> IO ()
testServer req _ respWriter = do
  case Server.requestPath req of
    Just "/inifite" -> do
      let infiniteBSWriter :: (Builder.Builder -> IO ()) -> IO () -> IO ()
          infiniteBSWriter bsWriter flush = do
            bsWriter $ Builder.lazyByteString $ LBS.concat $ replicate 1000 "foo\n"
            flush
            infiniteBSWriter bsWriter flush
          infiniteResponse = Server.responseStreaming status200 [] infiniteBSWriter
      respWriter infiniteResponse []
    _ -> do
      respWriter (Server.responseNoBody status404 []) []

testClient :: Client.Client ()
testClient client = client (Client.requestNoBody "GET" "/inifite" []) $ \res ->
  if Client.responseStatus res == Just status200
    then do
      bs <- Client.getResponseBodyChunk res
      putStrLn $ "Got chunk of size: " <> show (BS.length bs)
      threadDelay maxBound
    else error $ "the response isn't 200, due to a typo?"

main :: IO ()
main = do
  bracket (bindRandomPortTCP "*") (close . snd) $ \(serverPort, listenSock) -> do
    listen listenSock 1024
    _ <- async $ forever $ do
      (sock, _) <- accept listenSock
      let cleanup cfg = do
            Client.freeSimpleConfig cfg
            close sock
      async $ bracket (Server.allocSimpleConfig sock 4096) cleanup $ \cfg -> do
        Server.run cfg testServer

    let clientConfig =
          Client.ClientConfig
            { Client.scheme = "http",
              Client.authority = "localhost",
              Client.cacheLimit = 20
            }
    bracket (fst <$> getSocketTCP "localhost" serverPort) close $ \sock ->
      bracket (Client.allocSimpleConfig sock 4096) Client.freeSimpleConfig $ \http2Cfg -> do
        Client.run clientConfig http2Cfg $ testClient
    pure ()

I can see the traffic in wireshark, so this should mean that the server isn't stuck and filling up the memory. So, this has to be on the client side.