haskell-distributed · LaurentRDC · Aug 13, 2025
diff --git a/packages/network-transport-quic/LICENSE b/packages/network-transport-quic/LICENSE
@@ -0,0 +1,20 @@
+Copyright (c) Laurent P. René de Cotret
+
+Permission is hereby granted, free of charge, to any person obtaining
+a copy of this software and associated documentation files (the
+"Software"), to deal in the Software without restriction, including
+without limitation the rights to use, copy, modify, merge, publish,
+distribute, sublicense, and/or sell copies of the Software, and to
+permit persons to whom the Software is furnished to do so, subject to
+the following conditions:
+
+The above copyright notice and this permission notice shall be included
+in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
+IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
+CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
+TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
+SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
diff --git a/packages/network-transport-quic/network-transport-quic.cabal b/packages/network-transport-quic/network-transport-quic.cabal
@@ -0,0 +1,86 @@
+cabal-version: 3.0
+Name:          network-transport-quic
+Version:       0.1.0
+build-Type:    Simple
+License:       BSD-3-Clause
+License-file:  LICENSE
+Copyright:     Laurent P. René de Cotret
+Author:        Laurent P. René de Cotret
+maintainer:    The Distributed Haskell team
+Stability:     experimental
+Homepage:      http://haskell-distributed.github.com
+Bug-Reports:   https://github.com/haskell-distributed/distributed-process/issues
+Synopsis:      Networking layer for Cloud Haskell based on QUIC
+Description:   Networking layer for Cloud Haskell based on QUIC
+tested-with:   GHC==8.10.7 GHC==9.0.2 GHC==9.2.8 GHC==9.4.5 GHC==9.6.4 GHC==9.8.2 GHC==9.10.1 GHC==9.12.1
+Category:      Network
+extra-doc-files: ChangeLog
+extra-source-files: test/credentials/*
+
+source-repository head
+  Type:     git
+  Location: https://github.com/haskell-distributed/distributed-process
+  SubDir:   packages/network-transport-quic
+
+common common
+    ghc-options:
+                 -- warnings
+                 -Wall
+                 -Wcompat
+                 -Widentities
+                 -Wincomplete-uni-patterns
+                 -Wincomplete-record-updates
+                 -Wredundant-constraints
+                 -fhide-source-paths
+                 -Wpartial-fields
+                 -Wunused-packages
+                 -- The -threaded option is /required/ to use the quic library
+                 -threaded
+
+library
+  import:             common
+  build-depends:      attoparsec
+                    , base >= 4.14 && < 5
+                    , bytestring >= 0.10 && < 0.13
+                    , containers
+                    , ip
+                    , network >= 3.1 && < 3.3
+                    , network-transport >= 0.5 && < 0.6
+                    , quic ^>=0.2
+                    , stm >=2.4 && <2.6
+                    , text >= 2.0 && <2.2
+                    , tls
+                    , tls-session-manager
+  exposed-modules:    Network.Transport.QUIC
+                      Network.Transport.QUIC.Internal
+  other-modules:      Network.Transport.QUIC.Internal.Configuration
+                      Network.Transport.QUIC.Internal.EndpointState
+                      Network.Transport.QUIC.Internal.QUICAddr
+                      Network.Transport.QUIC.Internal.TLS
+                      Network.Transport.QUIC.Internal.TransportState
+  default-language:   Haskell2010
+  default-extensions: ImportQualifiedPost
+  -- The -threaded option is /required/ to use the quic library
+  hs-source-dirs:     src
+
+test-suite network-transport-quic-tests
+    import:             common
+    default-language:   Haskell2010
+    default-extensions: ImportQualifiedPost
+    main-is:            Main.hs
+    other-modules:      Test.Network.Transport.QUIC
+                        Test.Network.Transport.QUIC.Internal.QUICAddr
+    type:               exitcode-stdio-1.0
+    hs-source-dirs:     test
+    build-depends:      base
+                      , filepath
+                      , hedgehog
+                      , ip
+                      , network
+                      , network-transport
+                      , network-transport-quic
+                      , network-transport-tests
+                      , tasty ^>=1.5
+                      , tasty-hedgehog
+                      , tasty-hunit
+                      , text
diff --git a/packages/network-transport-quic/src/Network/Transport/QUIC.hs b/packages/network-transport-quic/src/Network/Transport/QUIC.hs
@@ -0,0 +1,16 @@
+module Network.Transport.QUIC (
+    createTransport,
+    QUICAddr (..),
+
+    -- * Re-export to generate credentials
+    Credential,
+    credentialLoadX509,
+) where
+
+import Network.Transport.QUIC.Internal (
+    -- \* Re-export to generate credentials
+    Credential,
+    QUICAddr (..),
+    createTransport,
+    credentialLoadX509,
+ )
diff --git a/packages/network-transport-quic/src/Network/Transport/QUIC/Internal.hs b/packages/network-transport-quic/src/Network/Transport/QUIC/Internal.hs
@@ -0,0 +1,208 @@
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+
+module Network.Transport.QUIC.Internal (
+    createTransport,
+    QUICAddr (..),
+    encodeQUICAddr,
+    decodeQUICAddr,
+
+    -- * Re-export to generate credentials
+    Credential,
+    credentialLoadX509,
+) where
+
+import Control.Concurrent (ThreadId, forkFinally, killThread, myThreadId)
+import Control.Concurrent.STM (atomically)
+import Control.Concurrent.STM.TQueue (
+    TQueue,
+    newTQueueIO,
+    readTQueue,
+    writeTQueue,
+ )
+import Control.Exception (bracket, try)
+import Control.Monad (void)
+import Data.Bifunctor (first)
+import Data.ByteString (StrictByteString)
+import Data.ByteString qualified as BS
+import Data.Foldable (traverse_)
+import Data.Functor (($>), (<&>))
+import Data.IORef (IORef, newIORef, readIORef, writeIORef)
+import Data.List.NonEmpty (NonEmpty)
+import Data.Set (Set)
+import Data.Set qualified as Set
+import GHC.IORef (atomicModifyIORef'_)
+import Network.QUIC (Stream)
+import Network.QUIC qualified as QUIC
+import Network.QUIC.Client qualified as QUIC.Client
+import Network.QUIC.Server qualified as QUIC.Server
+import Network.TLS (Credential)
+import Network.Transport (
+    ConnectErrorCode (ConnectNotFound),
+    ConnectHints,
+    Connection (..),
+    ConnectionId,
+    EndPoint (..),
+    EndPointAddress,
+    Event (..),
+    EventErrorCode (EventEndPointFailed),
+    NewEndPointErrorCode,
+    NewMulticastGroupErrorCode (NewMulticastGroupUnsupported),
+    Reliability,
+    ResolveMulticastGroupErrorCode (ResolveMulticastGroupUnsupported),
+    SendErrorCode (..),
+    Transport (..),
+    TransportError (..),
+ )
+import Network.Transport.QUIC.Internal.Configuration (credentialLoadX509, mkClientConfig, mkServerConfig)
+import Network.Transport.QUIC.Internal.QUICAddr (QUICAddr (..), decodeQUICAddr, encodeQUICAddr)
+import Network.Transport.QUIC.Internal.TransportState (TransportState, newTransportState, registerEndpoint, traverseTransportState)
+
+{- | Create a new Transport.
+
+Only a single transport should be created per Haskell process
+(threads can, and should, create their own endpoints though).
+-}
+createTransport ::
+    QUICAddr ->
+    NonEmpty Credential ->
+    IO Transport
+createTransport quicAddr creds = do
+    transportState <- newTransportState
+    pure $
+        Transport
+            (newEndpoint transportState quicAddr creds)
+            (closeQUICTransport transportState)
+
+newEndpoint ::
+    TransportState ->
+    QUICAddr ->
+    NonEmpty Credential ->
+    IO (Either (TransportError NewEndPointErrorCode) EndPoint)
+newEndpoint transportState quicAddr@(QUICAddr host port) creds = do
+    eventQueue <- newTQueueIO
+
+    state <- EndpointState <$> newIORef mempty
+
+    serverConfig <- mkServerConfig host port creds
+    serverThread <-
+        forkFinally
+            ( QUIC.Server.run
+                serverConfig
+                ( withQUICStream $
+                    -- TODO: create a bidirectional stream
+                    -- which can be re-used for sending
+                    \stream ->
+                        -- We register which threads are actively receiving or sending
+                        -- data such that we can cleanly stop
+                        withThreadRegistered state $ do
+                            -- TODO: how to ensure positivity of ConnectionId? QUIC StreamID should be a 62 bit integer,
+                            -- so there's room to make it a positive 64 bit integer (ConnectionId ~ Word64)
+                            let connId = fromIntegral (QUIC.streamId stream)
+                            receiveLoop connId stream eventQueue
+                )
+            )
+            ( \case
+                Left exc -> atomically (writeTQueue eventQueue (ErrorEvent (TransportError EventEndPointFailed (show exc))))
+                Right _ -> pure ()
+            )
+
+    let endpoint =
+            EndPoint
+                (atomically (readTQueue eventQueue))
+                (encodeQUICAddr quicAddr)
+                (connectQUIC creds)
+                (pure . Left $ TransportError NewMulticastGroupUnsupported "Multicast not supported")
+                (pure . Left . const (TransportError ResolveMulticastGroupUnsupported "Multicast not supported"))
+                (stopAllThreads state >> killThread serverThread >> atomically (writeTQueue eventQueue EndPointClosed))
+    void $ transportState `registerEndpoint` endpoint
+    pure $ Right endpoint
+  where
+    receiveLoop ::
+        ConnectionId ->
+        QUIC.Stream ->
+        TQueue Event ->
+        IO ()
+    receiveLoop connId stream eventQueue = do
+        incoming <- QUIC.recvStream stream 1024 -- TODO: variable length?
+        -- TODO: check some state whether we should stop all connections
+        if BS.null incoming
+            then do
+                atomically (writeTQueue eventQueue (ConnectionClosed connId))
+            else do
+                atomically (writeTQueue eventQueue (Received connId [incoming]))
+                receiveLoop connId stream eventQueue
+
+    withQUICStream :: (QUIC.Stream -> IO a) -> QUIC.Connection -> IO a
+    withQUICStream f conn =
+        bracket
+            (QUIC.waitEstablished conn >> QUIC.acceptStream conn)
+            (\stream -> QUIC.closeStream stream >> QUIC.Server.stop conn)
+            f
+
+connectQUIC ::
+    NonEmpty Credential ->
+    EndPointAddress ->
+    Reliability ->
+    ConnectHints ->
+    IO (Either (TransportError ConnectErrorCode) Connection)
+connectQUIC creds endpointAddress _reliability _connectHints =
+    case decodeQUICAddr endpointAddress of
+        Left errmsg -> pure $ Left $ TransportError ConnectNotFound ("Could not decode QUIC address: " <> errmsg)
+        Right (QUICAddr hostname port) ->
+            try $ do
+                clientConfig <- mkClientConfig hostname port creds
+
+                QUIC.Client.run clientConfig $ \conn -> do
+                    QUIC.waitEstablished conn
+                    stream <- QUIC.stream conn
+
+                    pure $
+                        Connection
+                            (sendQUIC stream)
+                            (QUIC.closeStream stream)
+  where
+    sendQUIC :: Stream -> [StrictByteString] -> IO (Either (TransportError SendErrorCode) ())
+    sendQUIC stream payloads =
+        try (QUIC.sendStreamMany stream payloads)
+            <&> first
+                ( \case
+                    QUIC.StreamIsClosed -> TransportError SendClosed "QUIC stream is closed"
+                    QUIC.ConnectionIsClosed reason -> TransportError SendClosed (show reason)
+                    other -> TransportError SendFailed (show other)
+                )
+
+closeQUICTransport :: TransportState -> IO ()
+closeQUICTransport = flip traverseTransportState (\_ endpoint -> closeEndPoint endpoint)
+
+{- | We keep track of all threads actively listening on QUIC streams
+so that we can cleanly stop these threads when closing the endpoint.
+
+See 'withThreadRegistered' for a combinator which automatically keeps
+track of these threads
+-}
+newtype EndpointState = EndpointState
+    { threads :: IORef (Set ThreadId)
+    }
+
+withThreadRegistered :: EndpointState -> IO a -> IO a
+withThreadRegistered state f =
+    bracket
+        registerThread
+        unregisterThread
+        (const f)
+  where
+    registerThread =
+        myThreadId
+            >>= \tid ->
+                atomicModifyIORef'_ (threads state) (Set.insert tid)
+                    $> tid
+
+    unregisterThread tid =
+        atomicModifyIORef'_ (threads state) (Set.insert tid)
+
+stopAllThreads :: EndpointState -> IO ()
+stopAllThreads (EndpointState tds) = do
+    readIORef tds >>= traverse_ killThread
+    writeIORef tds mempty -- so that we can call `closeQUICTransport` even after the endpoint has been closed
diff --git a/packages/network-transport-quic/src/Network/Transport/QUIC/Internal/Configuration.hs b/packages/network-transport-quic/src/Network/Transport/QUIC/Internal/Configuration.hs
@@ -0,0 +1,59 @@
+{-# LANGUAGE OverloadedStrings #-}
+
+module Network.Transport.QUIC.Internal.Configuration (
+    mkClientConfig,
+    mkServerConfig,
+
+    -- * Re-export to generate credentials
+    Credential,
+    TLS.credentialLoadX509,
+) where
+
+import Data.List.NonEmpty (NonEmpty)
+import Data.List.NonEmpty qualified as NonEmpty
+import Network.QUIC.Client (ClientConfig (ccALPN, ccValidate, ccWatchDog), ccPortName, ccServerName, defaultClientConfig)
+import Network.QUIC.Internal (ClientConfig (ccDebugLog), Milliseconds (Milliseconds), ServerConfig (scALPN), ccCredentials, ccKeyLog, maxIdleTimeout, scParameters)
+import Network.QUIC.Server (ServerConfig (scAddresses, scCredentials, scSessionManager, scUse0RTT), defaultServerConfig)
+import Network.Socket (HostName, ServiceName)
+import Network.TLS (Credential, Credentials (Credentials))
+import Network.Transport.QUIC.Internal.TLS qualified as TLS
+
+mkClientConfig ::
+    HostName ->
+    ServiceName ->
+    NonEmpty Credential ->
+    IO ClientConfig
+mkClientConfig host port creds = do
+    pure $
+        defaultClientConfig
+            { ccServerName = host
+            , ccPortName = port
+            , ccALPN = \_version -> pure (Just ["perf"])
+            , ccValidate = False
+            , ccCredentials = Credentials (NonEmpty.toList creds)
+            , ccWatchDog = True
+            , -- The following two parameters are for debugging. TODO: turn off by default
+              ccDebugLog = True
+            , ccKeyLog = putStrLn
+            }
+
+mkServerConfig ::
+    HostName ->
+    ServiceName ->
+    NonEmpty Credential ->
+    IO ServerConfig
+mkServerConfig host port creds = do
+    tlsSessionManager <- TLS.sessionManager
+
+    pure $
+        defaultServerConfig
+            { scAddresses = [(read host, read port)]
+            , scSessionManager = tlsSessionManager
+            , scCredentials = Credentials (NonEmpty.toList creds)
+            , scALPN = Just $ \_version _protocols -> pure "perf"
+            , scUse0RTT = True
+            , scParameters =
+                (scParameters defaultServerConfig)
+                    { maxIdleTimeout = Milliseconds 1000
+                    }
+            }