Channel.hs

-- | Channel
--
-- You should not have to import this module directly; instead import
-- "Network.GRPC.Util.Session".
module Network.GRPC.Util.Session.Channel (
    -- * Main definition
    Channel(..)
  , initChannel
    -- ** Flow state
  , RegularFlowState(..)
  , initFlowStateRegular
    -- * Working with an open channel
  , getInboundHeaders
  , send
  , recvBoth
  , recvEither
  , RecvFinal(..)
  , RecvAfterFinal(..)
  , SendAfterFinal(..)
    -- * Closing
  , waitForOutbound
  , close
  , ChannelDiscarded(..)
  , ChannelAborted(..)
    -- * Constructing channels
  , sendMessageLoop
  , recvMessageLoop
  , outboundTrailersMaker
  ) where


import Network.GRPC.Util.Imports

import Control.Concurrent.STM (STM, TVar, TMVar)
import Control.Concurrent.STM qualified as STM
import Data.ByteString.Builder (Builder)
import Data.ByteString.Lazy qualified as BS.Lazy

import Network.HTTP.Semantics qualified as HTTP.Semantics (
    TrailersMaker
  , NextTrailersMaker(..)
  )

import Network.GRPC.Common.Exception
import Network.GRPC.Common.StreamElem (StreamElem(..))
import Network.GRPC.Common.StreamElem qualified as StreamElem
import Network.GRPC.Spec.Util.Parser (Parser)
import Network.GRPC.Spec.Util.Parser qualified as Parser
import Network.GRPC.Util.RedundantConstraint
import Network.GRPC.Util.Session.API
import Network.GRPC.Util.Stream
import Network.GRPC.Util.Thread

{-------------------------------------------------------------------------------
  Definitions

  The fields of 'Channel' are its /implementation/, not its interface. It is
  kept opaque in the top-level @.Peer@ module.

  Implementation note: it is tempting to try and define 'Channel' purely in
  terms of bytestrings, and deal with serialization and deserialization to and
  from messages in a higher layer. However, this does not work:

  - For deserialization, if we make chunks of messages available in the 'TMVar',
    then if multiple threads are reading from that one 'TMVar', one thread might
    get the first chunk of a message and another thread the second.
  - Similarly, for serialization, if multiple threads are trying to write to the
    'TMVar', we might get interleaving of fragments of messages.

  Thus, to ensure thread safety, we work at the level of messages, not bytes.
-------------------------------------------------------------------------------}

-- | Bidirectional open channel on a node to a peer node
--
-- The node might be a client (and its peer a server), or the node might be
-- a server (and its peer a client); the main purpose of this abstraction
-- is precisely to abstract over that difference.
--
-- Each channel is constructed for a /single/ session (request/response).
data Channel sess = Channel {
      -- | Thread state of the thread receiving messages from the peer
      channelInbound :: TVar (FlowThreadState (Inbound sess))

      -- | Thread state of the thread sending messages to the peer
    , channelOutbound :: TVar (FlowThreadState (Outbound sess))

      -- | Have we sent the final message?
      --
      -- The sole purpose of this 'TVar' is catching user mistakes: if there is
      -- another 'send' after the final message, we can throw an exception,
      -- rather than the message simply being lost or blockng indefinitely.
    , channelSentFinal :: TVar (Maybe Backtraces)

      -- | Have we received the final message?
      --
      -- This is used to improve the user experience; see 'channelSentFinal'.
      -- It is also used when checking if a call should be considered
      -- \"cancelled\"; see 'withRPC'.
    , channelRecvFinal :: TVar (RecvFinal (Inbound sess))
    }

-- | Thread that deals with inbound or outbound flow
type FlowThreadState flow =
       ThreadState
         (RegularFlowState flow)
         (Trailers         flow)
         (NoMessages       flow)

-- | Interface to 'FlowThreadState'
type FlowThreadIface flow =
       ThreadIface
         (RegularFlowState flow)
         (NoMessages       flow)

-- | Has the client code received the final message from the peer yet?
--
-- NOTE: \"delivered\" here means: put the final message that we received from
-- the peer into the hands of the client code.
data RecvFinal flow =
    -- | We have not yet delivered the final message to the client code
    RecvNotFinal

    -- | We delivered the final message, but not yet the trailers
  | RecvWithoutTrailers (Trailers flow)

    -- | We delivered the final message and the trailers
  | RecvFinal Backtraces

deriving instance DataFlow flow => Show (RecvFinal flow)

-- | Regular (streaming) flow state
data RegularFlowState flow = RegularFlowState {
      -- | Headers
      --
      -- On the client side, the outbound headers are specified when the request
      -- is made ('callRequestMetadata'), and the inbound headers are recorded
      -- once the responds starts to come in; clients can block-and-wait for
      -- these headers ('getInboundHeaders').
      --
      -- On the server side, the inbound headers are recorded when the request
      -- comes in, and the outbound headers are specified
      -- ('setResponseInitialMetadata') before the response is initiated
      -- ('initiateResponse'/'sendTrailersOnly').
      flowHeaders :: Headers flow

      -- | Messages
      --
      -- This TMVar is written to for incoming messages ('recvMessageLoop') and
      -- read from for outgoing messages ('sendMessageLoop'). It acts as a
      -- one-place buffer, providing backpressure in both directions.
    , flowMsg :: TMVar (StreamElem (Trailers flow) (Message flow))

      -- | Trailers
      --
      -- Unlike 'flowMsg', which is /written/ to in 'recvMessageLoop' and /read/
      -- from in 'sendMessageLoop', both loops /set/ 'flowTerminated', once,
      -- just before they terminate.
      --
      -- * For 'sendMessageLoop', this means that the last message has been
      --   written (that is, the last call to 'writeChunk' has happened).
      --   This has two consequences:
      --
      --   1. @http2@ can now construct the trailers ('outboundTrailersMaker')
      --   2. Higher layers can wait on 'flowTerminated' to be /sure/ that the
      --      last message has been written.
      --
      -- * For 'recvMessageLoop', this means that the trailers have been
      --   received from the peer. Higher layers can use this to check for, or
      --   block-and-wait, to receive those trailers.
      --
      -- == Relation to 'channelSentFinal'/'channelRecvFinal'
      --
      -- 'flowTerminated' is set at different times than 'channelSentFinal' and
      -- 'channelRecvFinal' are:
      --
      -- * 'channelSentFinal' is set on the last call to 'send', but /before/
      --   the message is processed by 'sendMessageLoop'.
      -- * 'channelRecvFinal', dually, is set on the last call to 'recv, which
      --   must (necessarily) happen /before/ that message is actually made
      --   available by 'recvMessageLoop'.
      --
      -- /Their/ sole purpose is to catch user errors, not capture data flow.
      --
      -- == Relation to 'ThreadState'
      --
      -- Although the threads write their final result (that is, the trailers)
      -- to the 'ThreadState', we cannot use that in the trailers maker, because
      -- the trailers are constructed /within/ the thread: that is, before it
      -- terminates.
    , flowTerminated :: TMVar (Trailers flow)
    }

-- | 'Show' instance is useful in combination with @stm-debug@ only
deriving instance (
    Show (Headers flow)
  , Show (TMVar (StreamElem (Trailers flow) (Message flow)))
  , Show (TMVar (Trailers flow))
  ) => Show (RegularFlowState flow)

{-------------------------------------------------------------------------------
  Initialization
-------------------------------------------------------------------------------}

initChannel ::
     String
     -- ^ Role (server or client)
     --
     -- This is used for debugging, to label the inbound and outbound thread.
  -> IO (Channel sess)
initChannel role = do
    channelInbound   <- newThreadState (role ++ "/inbound")
    channelOutbound  <- newThreadState (role ++ "/outbound")
    channelSentFinal <- STM.newTVarIO Nothing
    channelRecvFinal <- STM.newTVarIO RecvNotFinal
    return Channel{
        channelInbound
      , channelOutbound
      , channelSentFinal
      , channelRecvFinal
      }

initFlowStateRegular :: Headers flow -> IO (RegularFlowState flow)
initFlowStateRegular flowHeaders = do
   flowMsg        <- STM.newEmptyTMVarIO
   flowTerminated <- STM.newEmptyTMVarIO
   return RegularFlowState {
       flowHeaders
     , flowMsg
     , flowTerminated
     }

{-------------------------------------------------------------------------------
  Working with an open channel
-------------------------------------------------------------------------------}

-- | The inbound headers
--
-- Will block if the inbound headers have not yet been received.
getInboundHeaders ::
     Channel sess
  -> IO (Either (NoMessages (Inbound sess)) (Headers (Inbound sess)))
getInboundHeaders Channel{channelInbound} =
    withThreadInterface channelInbound (return . aux)
  where
    aux :: FlowThreadIface flow -> Either (NoMessages flow) (Headers flow)
    aux = \case
      IfaceAvailable regular  -> Right $ flowHeaders regular
      IfaceTrivial   trailers -> Left trailers

-- | Send a message to the node's peer
--
-- It is a bug to call 'send' again after the final message (that is, a message
-- which 'StreamElem.whenDefinitelyFinal' considers to be final). Doing so will
-- result in a 'SendAfterFinal' exception.
send :: forall sess.
     (HasCallStack, NFData (Message (Outbound sess)))
  => Channel sess
  -> StreamElem (Trailers (Outbound sess)) (Message (Outbound sess))
  -> IO ()
send Channel{channelOutbound, channelSentFinal} = \msg -> do
    msg' <- evaluate $ force <$> msg
    backtrace <- collectBacktraces
    withThreadInterface channelOutbound $ aux backtrace msg'
  where
    aux ::
         Backtraces
      -> StreamElem (Trailers (Outbound sess)) (Message (Outbound sess))
      -> FlowThreadIface (Outbound sess)
      -> STM ()
    aux backtrace msg iface = do
        -- By checking that we haven't sent the final message yet, we know that
        -- this call to 'putMVar' will not block indefinitely: the thread that
        -- sends messages to the peer will get to it eventually (unless it dies,
        -- in which case the thread status will change and the call to
        -- 'getThreadInterface' will be retried).
        sentFinal <- STM.readTVar channelSentFinal
        case sentFinal of
          Just cs -> STM.throwSTM $ SendAfterFinal cs
          Nothing -> return ()
        case iface of
          IfaceAvailable regular -> do
            StreamElem.whenDefinitelyFinal msg $ \_trailers ->
              STM.writeTVar channelSentFinal $ Just backtrace
            STM.putTMVar (flowMsg regular) msg
          IfaceTrivial _trailers ->
            -- For outgoing messages, the caller decides to use Trailers-Only,
            -- so if they then subsequently call 'send', we throw an exception.
            -- This is different for /inbound/ messages; see 'recv', below.
            STM.throwSTM $ SendButTrailersOnly

-- | Receive a message from the node's peer
--
-- If the sender indicates that the message is final /when/ they send it, by
-- sending the HTTP trailers in the same frame, then we will return the message
-- and the trailers together. It is a bug to call 'recvBoth' again after this;
-- doing so will result in a 'RecvAfterFinal' exception.
recvBoth :: forall sess.
     (HasCallStack, IsSession sess)
  => Channel sess
  -> IO ( Either
            (NoMessages (Inbound sess))
            (StreamElem (Trailers (Inbound sess)) (Message (Inbound sess)))
        )
recvBoth =
    recv'
      StreamElem
      NoMoreElems
      ((,Nothing) . uncurry FinalElem)

-- | Variant on 'recvBoth' where trailers are always returned separately
--
-- Unlike in 'recvBoth', even if the sender indicates that the final message is
-- final when they send it, we will store these trailers internally and return
-- only that final message. The trailers are then returned on the /next/ call to
-- 'recvEither'. Call 'recvEither' again /after/ receiving the trailers is a
-- bug; doing so will result in a 'RecvAfterFinal' exception.
recvEither ::
     (HasCallStack, IsSession sess)
  => Channel sess
  -> IO ( Either
            (NoMessages (Inbound sess))
            (Either (Trailers (Inbound sess)) (Message (Inbound sess)))
        )
recvEither =
    recv'
      Right
      Left
      (bimap Right Just)

-- | Internal generalization of 'recvBoth' and 'recvEither'
recv' :: forall sess b.
     (HasCallStack, IsSession sess)
  => (Message    (Inbound sess) -> b)  -- ^ Message without trailers
  -> (Trailers   (Inbound sess) -> b)  -- ^ Trailers without (final) message
  -> (    (Message (Inbound sess), Trailers (Inbound sess))
       -> (b, Maybe (Trailers (Inbound sess)))
     )
     -- ^ Message with trailers
     --
     -- In addition to the result, should also return the trailers to keep for
     -- the next call to 'recv'' (if any).
  -> Channel sess
  -> IO (Either (NoMessages (Inbound sess)) b)
recv' messageWithoutTrailers
      trailersWithoutMessage
      messageWithTrailers
      Channel{channelInbound, channelRecvFinal} = do
    backtrace <- collectBacktraces
    withThreadInterface channelInbound $ aux backtrace
  where
    _ = addConstraint @(IsSession sess)

    aux ::
         Backtraces
      -> FlowThreadIface (Inbound sess)
      -> STM (Either (NoMessages (Inbound sess)) b)
    aux backtrace iface = do
        -- By checking that we haven't received the final message yet, we know
        -- that this call to 'takeTMVar' will not block indefinitely: the thread
        -- that receives messages from the peer will get to it eventually
        -- (unless it dies, in which case the thread status will change and the
        -- call to 'getThreadInterface' will be retried).
        readFinal <- STM.readTVar channelRecvFinal
        case readFinal of
          RecvNotFinal ->
            case iface of
              IfaceAvailable regular -> Right <$> do
                streamElem <- STM.takeTMVar (flowMsg regular)
                -- We update 'channelRecvFinal' in the same tx as the read, to
                -- atomically change "there is a value" to "all values read".
                case streamElem of
                  StreamElem msg ->
                    return $ messageWithoutTrailers msg
                  FinalElem msg trailers -> do
                    let (b, mTrailers) = messageWithTrailers (msg, trailers)
                    STM.writeTVar channelRecvFinal $
                      maybe (RecvFinal backtrace) RecvWithoutTrailers mTrailers
                    return $ b
                  NoMoreElems trailers -> do
                    STM.writeTVar channelRecvFinal $ RecvFinal backtrace
                    return $ trailersWithoutMessage trailers
              IfaceTrivial trailers -> do
                STM.writeTVar channelRecvFinal $ RecvFinal backtrace
                return $ Left trailers
          RecvWithoutTrailers trailers -> do
            STM.writeTVar channelRecvFinal $ RecvFinal backtrace
            return $ Right $ trailersWithoutMessage trailers
          RecvFinal cs ->
            STM.throwSTM $ RecvAfterFinal cs

-- | Thrown by 'send'
--
-- See 'send' for additional discussion.
data SendAfterFinal =
    -- | Call to 'send' after the final message was sent
    --
    -- We record the backtrace of final call to 'send'.
    SendAfterFinal Backtraces

    -- | Call to 'send', but we are in the Trailers-Only case
  | SendButTrailersOnly
  deriving stock (Show)
  deriving anyclass (Exception)

-- | Thrown by 'recv'
--
--
-- See 'recv' for additional discussion.
data RecvAfterFinal =
     -- | Call to 'recv' after the final message was already received
     --
     -- We record the backtrace of final call to 'recv'.
     RecvAfterFinal Backtraces
  deriving stock (Show)
  deriving anyclass (Exception)

{-------------------------------------------------------------------------------
  Closing
-------------------------------------------------------------------------------}

-- | Wait for the outbound thread to terminate
--
-- See 'close' for discussion.
waitForOutbound :: HasCallStack => Channel sess -> IO ()
waitForOutbound Channel{channelOutbound} =
    void $ waitForNormalThreadTermination channelOutbound

-- | Close the channel
--
-- Before a channel can be closed, you should 'send' the final outbound message
-- and then 'waitForOutbound' until all outbound messages have been processed.
-- It is not possible to do this implicitly, because the final call to 'send'
-- involves a choice of trailers, and calling 'waitForOutbound' /without/ a
-- final close to 'send' will result in deadlock. Typically code will also
-- process all /incoming/ messages, but doing so is of course not mandatory.
--
-- If the outbound thread is still running, 'waitForOutbound' was
-- not called, and the outbound thread will be terminated with an exception:
--
-- * If the channel is closed /because of/ an exception, we use that exception
--   (or 'ChannelAborted' in the case of 'ExitCaseAbort')
-- * Otherwise, the caller terminated normally and yet did not call
--   'waitForOutbound'. This is a bug in the caller, which we record as a
--   'ChannelDiscarded' exception on the channel.
close ::
     HasCallStack
  => Channel sess
  -> ExitCase a    -- ^ The reason why the channel is being closed
  -> IO ()
close Channel{channelOutbound} reason = do
    backtrace <- collectBacktraces
    let channelClosed :: ExactException
        channelClosed = WrapExactException $
            case reason of
              ExitCaseSuccess _   -> toException $ ChannelDiscarded backtrace
              ExitCaseAbort       -> toException $ ChannelAborted   backtrace
              ExitCaseException e -> e

    -- 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.
    cancelThread channelOutbound channelClosed

-- | Channel was closed because it was discarded
--
-- This typically corresponds to leaving the scope of 'runHandler' or
-- 'withRPC' (without throwing an exception).
data ChannelDiscarded = ChannelDiscarded Backtraces
  deriving stock (Show)
  deriving anyclass (Exception)

-- | Channel was closed for an unknown reason
--
-- This will only be used in monad stacks that have error mechanisms other
-- than exceptions.
data ChannelAborted = ChannelAborted Backtraces
  deriving stock (Show)
  deriving anyclass (Exception)

{-------------------------------------------------------------------------------
  Constructing channels

  Both 'sendMessageLoop' and 'recvMessageLoop' will be run in newly forked
  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 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'.
-------------------------------------------------------------------------------}

-- | Send all messages to the node's peer
--
-- == Invariant: eventual progress
--
-- The outbound thread, the thread running `sendMessageLoop`, communicates by
-- reading a shared 'TMVar'. When a write to this 'TMVar' is blocked (in another
-- thread), eventually one of the following two things will happen:
--
-- * The outbound thread empties the 'TMVar'
-- * The outbound thread dies (itself observable;
--   see 'Network.GRPC.Util.Thread.withThreadInterface').
--
-- This invariant will cease to be true after the final message ('FinalElem' or
-- 'NoMoreElems') is consumed: the thread will not read from the 'TMVar' after
-- that point.
--
-- == Exceptions
--
-- When a write /fails/ (say, connection lost) we'd like to be able to
-- communicate this back to the caller: the outbound thread dies, which is
-- observable (see above). It is important to note that the /absence/ of such an
-- exception (a quote-unquote \"successful\" write) is no guarantee of anything;
-- for example, it may be that the connection is lost after the message was
-- successfully enqueued in some OS buffer but before it was put on the wire; or
-- indeed somewhere along the way across the network to the destination.
--
-- However, the outbound thread spends most of its time blocked waiting for
-- messages to send to the network peer, and may not notice when the connection
-- is lost. It therefore /monitors/ the inbound thread, which spends most of its
-- time blocked on waiting for messages /from/ the peer and so will notice more
-- or less immediately.
--
-- We need an important \"atomicity\" property however: once the outbound thread
-- has sent the trailers, we /expect/ the client to disconnect. We therefore
-- mark ourselves as done prior to sending the trailers, to avoid a race
-- condition where
--
-- 1. the outbound thread sends the final chunk
-- 2. the client receives the trailers and disconnects
-- 3. before the outbound thread gets the chance to terminate (the only thing
--    left to do), it is killed (perhaps due to a monitor notification, or due
--    to @http2@ sending an async exception to a server handler)
--
-- Such a race condition would result in timing-sensitive, non-deterministic
-- exceptions. By marking ourselves done, /we cannot be killed anymore/, hence
-- preventing the problem.
--
-- Note that if a client disconnects /before/ receiving the final chunk, this
-- constitutes a violation of the protocol, and so it would be correct for the
-- outbound thread to report abnormal termination.
--
-- == Failure on the final message
--
-- Conceptually, we'd want something like
--
-- > do ..
-- >    writeChunkFinal ..
-- >    -- .. prevent async exceptions here ..
--
-- but of course that is impossible to do /literally/; anything we do /after/
-- the call to 'writeChunkFinal' still leaves a gap in between 'writeChunkFinal'
-- and the next instruction. However, we cannot mask async exceptions /before/
-- the call to 'writeChunkFinal' either, because 'writeChunkFinal' itself may
-- block, and if it does, we do want to be interruptible while we wait.
--
-- By declaring ourselves done /before/ sending the final chunk (see previous
-- section) we side-step the problem, at the cost of being unable to report if
-- that final send fails. However, this is a small price to pay:
--
-- * As discussed above, the absence of a reported failure of a send is /anyway/
--   no guarantee of success
-- * Reporting failed sends is primarily useful for code that is repeatedly
--   sending messages, and so if one message fails to send there is no point in
--   sending the next. But for the final message there /cannot be/ a next
--   message: this must anyway be the final write.
sendMessageLoop :: forall sess.
     IsSession sess
  => sess
  -> RegularFlowState (Outbound sess)
  -> OutputStream
  -> (Trailers (Outbound sess) -> IO ())
  -> IO ()
sendMessageLoop sess st stream markDone = do
    trailers <- loop
    atomically $ STM.putTMVar (flowTerminated st) trailers
  where
    build :: (Message (Outbound sess) -> Builder)
    build = buildMsg sess (flowHeaders st)

    loop :: IO (Trailers (Outbound sess))
    loop = do
        msg <- atomically $ STM.takeTMVar (flowMsg st)
        case msg of
          StreamElem x -> do
            writeChunk stream $ build x
            flush stream
            loop
          FinalElem x trailers -> do
            markDone trailers
            writeChunkFinal stream $ build x
            return trailers
          NoMoreElems trailers -> do
            markDone trailers
            -- Send empty chunk marked \"final\" to let our peer know that we
            -- have sent our last message. Note that this does not /necessarily/
            -- write a DATA frame, since http2 avoids writing empty data frames
            -- unless they are marked @END_OF_STREAM@.
            writeChunkFinal stream $ mempty
            return trailers

-- | Receive all messages sent by the node's peer
recvMessageLoop :: forall sess.
     (IsSession sess, HasCallStack)
  => sess
  -> RegularFlowState (Inbound sess)
  -> InputStream
  -> IO (Trailers (Inbound sess))
recvMessageLoop sess st stream =
    go $ parseMsg sess (flowHeaders st)
  where
    go :: Parser String (Message (Inbound sess)) -> IO (Trailers (Inbound sess))
    go parser = do
        mProcessedFinal <- throwParseErrors =<< Parser.processAll
          (getChunk stream)
          processOne
          processFinal
          parser
        case mProcessedFinal of
          Just trailers ->
            return trailers
          Nothing -> do
            trailers <- processTrailers
            atomically $ STM.putTMVar (flowMsg st) $ NoMoreElems trailers
            return trailers

    processOne :: Message (Inbound sess) -> IO ()
    processOne msg = do
        atomically $ STM.putTMVar (flowMsg st) $ StreamElem msg

    processFinal :: Message (Inbound sess) -> IO (Trailers (Inbound sess))
    processFinal msg = do
        trailers <- processTrailers
        atomically $ STM.putTMVar (flowMsg st) $ FinalElem msg trailers
        return trailers

    processTrailers :: IO (Trailers (Inbound sess))
    processTrailers = do
        trailers <- parseInboundTrailers sess =<< getTrailers stream
        atomically $ STM.putTMVar (flowTerminated st) $ trailers
        return trailers

    throwParseErrors :: Parser.ProcessResult String b -> IO (Maybe b)
    throwParseErrors (Parser.ProcessError err) =
        throwIO $ PeerSentMalformedMessage err
    throwParseErrors (Parser.ProcessedWithFinal b leftover) = do
        unless (BS.Lazy.null leftover) $ throwIO PeerSentIncompleteMessage
        return $ Just b
    throwParseErrors (Parser.ProcessedWithoutFinal leftover) = do
        unless (BS.Lazy.null leftover) $ throwIO PeerSentIncompleteMessage
        return $ Nothing

outboundTrailersMaker :: forall sess.
     IsSession sess
  => sess
  -> Channel sess
  -> RegularFlowState (Outbound sess)
  -> HTTP.Semantics.TrailersMaker
outboundTrailersMaker sess Channel{channelOutbound} regular = go
  where
    go :: HTTP.Semantics.TrailersMaker
    go (Just _) = return $ HTTP.Semantics.NextTrailersMaker go
    go Nothing  = do
        mFlowState <- atomically $
          unlessAbnormallyTerminated channelOutbound $
            STM.readTMVar (flowTerminated regular)
        case mFlowState of
            Right trailers ->
              return $ HTTP.Semantics.Trailers $ buildOutboundTrailers sess trailers
            Left _exception ->
              return $ HTTP.Semantics.Trailers []