quic 0.3.19 → 0.3.20

raw patch · 54 files changed

+2710/−1034 lines, 54 filesdep ~networkPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependency ranges changed: network

API changes (from Hackage documentation)

Files

• ChangeLog.md +61/−1
• Network/QUIC.hs +8/−0
• Network/QUIC/Client/Reader.hs +54/−26
• Network/QUIC/Client/Run.hs +49/−41
• Network/QUIC/Client/WatchDog.hs +53/−0
• Network/QUIC/Closer.hs +12/−3
• Network/QUIC/Connection.hs +2/−0
• Network/QUIC/Connection/Migration.hs +20/−6
• Network/QUIC/Connection/Misc.hs +80/−5
• Network/QUIC/Connection/PacketNumber.hs +6/−7
• Network/QUIC/Connection/Queue.hs +23/−15
• Network/QUIC/Connection/RateControl.hs +132/−0
• Network/QUIC/Connection/State.hs +15/−9
• Network/QUIC/Connection/Stream.hs +28/−4
• Network/QUIC/Connection/StreamTable.hs +94/−62
• Network/QUIC/Connection/Types.hs +88/−14
• Network/QUIC/Connection/Types.hs-boot +4/−0
• Network/QUIC/Handshake.hs +36/−4
• Network/QUIC/IO.hs +65/−58
• Network/QUIC/Internal.hs +2/−0
• Network/QUIC/Packet/Decrypt.hs +4/−1
• Network/QUIC/Packet/Encode.hs +8/−8
• Network/QUIC/Packet/Frame.hs +10/−5
• Network/QUIC/Receiver.hs +299/−220
• Network/QUIC/Recovery/Timer.hs +10/−6
• Network/QUIC/Sender.hs +47/−9
• Network/QUIC/Server/Reader.hs +41/−64
• Network/QUIC/Server/Run.hs +52/−53
• Network/QUIC/Socket.hs +9/−4
• Network/QUIC/Stream.hs +9/−4
• Network/QUIC/Stream/Frag.hs +0/−41
• Network/QUIC/Stream/Misc.hs +66/−56
• Network/QUIC/Stream/Queue.hs +4/−1
• Network/QUIC/Stream/Reass.hs +58/−79
• Network/QUIC/Stream/Skew.hs +0/−91
• Network/QUIC/Stream/Types.hs +42/−36
• Network/QUIC/Types.hs +2/−0
• Network/QUIC/Types/Ack.hs +18/−0
• Network/QUIC/Types/Delay.hs +35/−0
• Network/QUIC/Types/Exception.hs +1/−0
• Network/QUIC/Types/Frame.hs +1/−11
• Network/QUIC/Types/Packet.hs +4/−0
• cbits/sysevent.c +16/−5
• quic.cabal +8/−6
• test/Config.hs +102/−53
• test/HandshakeSpec.hs +106/−0
• test/IOSpec.hs +641/−4
• test/MigrationSpec.hs +87/−2
• test/PacketSpec.hs +44/−0
• test/ReassSpec.hs +38/−1
• test/ResetSpec.hs +17/−7
• test/TransportError.hs +36/−12
• test/TypesSpec.hs +31/−0
• test/WatchDogSpec.hs +32/−0

@@ -1,6 +1,66 @@ # ChangeLog  +## 0.3.20++Requires network 3.2.12, and allows tls 2.5.  `closeStream` and+`resetStream` now send STOP_SENDING while the peer has not finished the+stream, and `closeStreamWith` gives its error code.  `abortConnection`+returns at once instead of waiting for the exception to be taken.+`Network.QUIC.Internal.rebind` loses its `Microseconds` argument, and+the `Delay` of an ACK frame is a newtype of its own, with `noDelay`.++* Keep held stream fragments disjoint+  [#179](https://github.com/kazu-yamamoto/quic/pull/179)+* Give RESET_STREAM the final size of what was queued+  [#180](https://github.com/kazu-yamamoto/quic/pull/180)+* Scale ACK Delay by the peer's ack_delay_exponent+  [#181](https://github.com/kazu-yamamoto/quic/pull/181)+* Cap the size of ACK frames+  [#182](https://github.com/kazu-yamamoto/quic/pull/182)+* Discard Version Negotiation and Retry once a packet is processed+  [#183](https://github.com/kazu-yamamoto/quic/pull/183)+* Fail version negotiation when no version is in common+  [#184](https://github.com/kazu-yamamoto/quic/pull/184)+* Keep the dispatcher receiving after any number of failed receives+  [#185](https://github.com/kazu-yamamoto/quic/pull/185)+* Answer every STOP_SENDING and NEW_CONNECTION_ID+  [#186](https://github.com/kazu-yamamoto/quic/pull/186)+* Honor the peer's max_udp_payload_size+  [#187](https://github.com/kazu-yamamoto/quic/pull/187)+* Decode a packet number against the largest in its own space+  [#188](https://github.com/kazu-yamamoto/quic/pull/188)+* Close with FRAME_ENCODING_ERROR on a frame cut short+  [#189](https://github.com/kazu-yamamoto/quic/pull/189)+* Close the socket rebind replaced even if the connection ends first+  [#190](https://github.com/kazu-yamamoto/quic/pull/190)+* Close the socket rebind makes if its connect fails+  [#191](https://github.com/kazu-yamamoto/quic/pull/191)+* Close the client's socket when the peer ends the connection+  [#192](https://github.com/kazu-yamamoto/quic/pull/192)+* Stop the watch dog instead of spinning when its socket fails+  [#193](https://github.com/kazu-yamamoto/quic/pull/193)+* Let the client honor a CONNECTION_CLOSE with NO_ERROR+  [#194](https://github.com/kazu-yamamoto/quic/pull/194)+* Make a stateless reset smaller than the packet it answers+  [#195](https://github.com/kazu-yamamoto/quic/pull/195)+* Move the server to a new path only for a packet that decrypts+  [#196](https://github.com/kazu-yamamoto/quic/pull/196)+* Give up a path validation in time and go back to the old path+  [#197](https://github.com/kazu-yamamoto/quic/pull/197)+* Bound the streams kept after the application closes them+  [#198](https://github.com/kazu-yamamoto/quic/pull/198)+* Close our sockets only where no thread is inside a call on them+  [#199](https://github.com/kazu-yamamoto/quic/pull/199)+* Count received stream data by the largest offset+  [#200](https://github.com/kazu-yamamoto/quic/pull/200)+* Take a stream out of the table and note its frames in one transaction each+  [#201](https://github.com/kazu-yamamoto/quic/pull/201)+* Use remembered limits only for 0-RTT and refuse ones lowered under it+  [#202](https://github.com/kazu-yamamoto/quic/pull/202)+* Ask a connection to abort instead of throwing to its main thread+  [#203](https://github.com/kazu-yamamoto/quic/pull/203)+ ## 0.3.19  * The dispatcher is stopped by the stop variable alone: it waits for that@@ -15,7 +75,7 @@ * The dispatcher's receive tells a closed socket from a datagram's   failure with network's `isClosedSocketError` rather than by comparing   the error type here.-* Requires network 3.2.10 and network-run 0.7.+* Requires network 3.2.10. * The closer's drain gives up on its receive at the deadline rather than   having `timeout` abandon it, so the six calls it could leave behind, and   the close which ended them, are gone.

@@ -31,6 +31,7 @@      -- ** Closing     closeStream,+    closeStreamWith,     shutdownStream,     resetStream,     stopStream,@@ -93,6 +94,13 @@     ),     cryptoError,     ApplicationProtocolError (..),++    -- * Rate control+    PerSecond,+    pathAndCIDFrameLimit,+    pathResponseLimit,+    queuedControlLimit,+    statelessResetLimit, ) where  import Network.QUIC.Connection

@@ -15,8 +15,9 @@ import Data.Functor (($>)) import qualified Control.Monad.STM as STM import qualified Control.Exception as E+import qualified Data.ByteString as BS import Data.List (intersect)-import Network.Socket (Socket, close, connect, getSocketName)+import Network.Socket (Socket, getSocketName) import qualified Network.Socket.ByteString as NSB  import Network.QUIC.Common@@ -40,13 +41,12 @@ --   replaced, and when the connection has been idle for too long. -- --   Nothing is closed and nothing is thrown to make it return: the receive---   and both of those are waited for at once.  The socket it was reading is---   the caller's to close afterwards -- __not at once, though__: nothing---   here says when this has returned, so a caller which closes before it---   has is back to closing a socket under a reader.  'rebind' leaves the---   delay it is given before closing, for that.+--   and both of those are waited for at once.  A socket retired by 'rebind'+--   is left for the sender here rather than closed by the caller, which+--   cannot see when this has returned and would be closing it under a+--   reader. readerClient :: TVar Bool -> Socket -> Connection -> IO ()-readerClient retired s0 conn = handleLogUnit logAction $ do+readerClient retired s0 conn = handleLogUnit logAction $ (`E.finally` leave) $ do     labelMe "readerClient"     wait     connected <- getSockConnected conn@@ -62,6 +62,16 @@                         | otherwise -> return $ Right bs     loop recv   where+    -- A socket 'rebind' swapped out is nobody's but the retired list's,+    -- so it is left there however this thread ends -- told, thrown at or+    -- failed.  One still the connection's belongs to the closer, which+    -- reads and sends on it after this.+    --+    -- Left rather than closed because the sender may be inside a send on+    -- it; it closes them between two sends.+    leave = do+        current <- getSocketRetire conn+        when (current /= retired) $ retireSocket conn s0     wait = do         bound <- E.handle (throughAsync (return False)) $ do             _ <- getSocketName s0@@ -77,9 +87,13 @@                 ((readTVar retired >>= STM.check) $> Retired)                     `orElse` ((readTVar expired >>= STM.check) $> TimedOut)         case ebs of-            -- The socket belongs to whoever retired it; it closes it.+            -- 'leave' has it.             Left Retired -> return ()-            Left TimedOut -> close s0+            -- What ends the connection on an idle timeout is the+            -- receiver throwing 'ConnectionIsTimeout'; this thread only+            -- stops reading.  The socket goes the same way as a retired+            -- one, because the sender may be inside a send on it.+            Left TimedOut -> retireSocket conn s0             Right bs -> do                 now <- getTimeMicrosecond                 let quicBit = greaseQuicBit $ getMyParameters conn@@ -93,12 +107,18 @@         myVerInfo <- getVersionInfo conn         let myVer = chosenVersion myVerInfo             myVers0 = otherVersions myVerInfo-        -- ignoring VN if the original version is included.-        when (myVer `notElem` peerVers && Negotiation `notElem` peerVers) $ do-            ok <- checkCIDs conn dCID (Left sCID)+        -- RFC 9000 Sec 6.2: discarded once any other packet, an earlier+        -- Version Negotiation included, has been processed, and if it+        -- lists the version we chose.+        done <- processed conn+        icvn <- getIncompatibleVN conn+        ok <- checkCIDs conn dCID (Left sCID)+        let late = done || icvn+            listed = myVer `elem` peerVers || Negotiation `elem` peerVers+        when (ok && not late && not listed) $ do             let myVers = filter (not . isGreasingVersion) myVers0                 nextVerInfo = case myVers `intersect` peerVers of-                    vers@(ver : _) | ok -> VersionInfo ver vers+                    vers@(ver : _) -> VersionInfo ver vers                     _ -> brokenVersionInfo             E.throwTo (mainThreadId conn) $ VerNego nextVerInfo     putQ t (PacketIC pkt@(CryptPacket hdr crypt) lvl siz) = do@@ -121,8 +141,11 @@                 _ -> return () -- really invalid, just ignore     putQ t (PacketIR pkt@(RetryPacket ver dCID sCID token ex)) = do         qlogReceived conn pkt t+        -- RFC 9000 Sec 17.2.5.2: at most one Retry, none after an Initial+        -- has been processed, and none with an empty token.+        late <- processed conn         ok <- checkCIDs conn dCID ex-        when ok $ do+        when (ok && not late && not (BS.null token)) $ do             -- Re-creating peer's CIDDB with peer's CID.             resetPeerCID conn sCID             setPeerAuthCIDs conn $ \auth -> auth{retrySrcCID = Just sCID}@@ -133,6 +156,13 @@       where         put ppkt = putOutput conn $ OutRetrans ppkt +-- | Whether a packet from the server has been decrypted or a Retry taken.+processed :: Connection -> IO Bool+processed conn = do+    rx <- getRxBytes conn+    retried <- getRetried conn+    return (rx > 0 || retried)+ checkCIDs :: Connection -> CID -> Either CID (ByteString, ByteString) -> IO Bool checkCIDs conn dCID (Left sCID) = do     localCID <- getMyCID conn@@ -230,7 +260,7 @@ -- So -> Co: PathChallenge (validatePath) -- Co -> So: PathResponse (processFrame) controlConnection' conn NATRebinding = do-    rebind conn $ Microseconds 5000 -- nearly 0+    rebind conn     return True ---------------------------------------------------------------- -- ActiveMigration (-A)@@ -248,27 +278,25 @@         Nothing -> return False         mcidinfo -> do             peersa <- peerSockAddr <$> getPathInfo conn-            rebind conn $ Microseconds 5000000+            rebind conn             -- Sending PathChallenge, RetireConnectionID and NewConnectionID RPT             pathInfo <- newPathInfo peersa             addPathInfo conn pathInfo             validatePath conn pathInfo mcidinfo-            return True -rebind :: Connection -> Microseconds -> IO ()-rebind conn microseconds = do+rebind :: Connection -> IO ()+rebind conn = do     peersa <- peerSockAddr <$> getPathInfo conn-    newSock <- natRebinding peersa     connected <- getSockConnected conn-    when connected $ connect newSock peersa+    newSock <- natRebinding connected peersa     newRetired <- newTVarIO False-    (oldSock, oldRetired) <- swapSocket conn newSock newRetired+    (_, oldRetired) <- swapSocket conn newSock newRetired     (forkManaged conn "QUIC reader" $ readerClient newRetired newSock conn)         `E.finally` do             -- Telling the old reader to go rather than closing the socket             -- under it: a close ends a thread blocked on a descriptor only-            -- because the event manager says so, and a descriptor number is-            -- reused.  Said even if the fork above failed, since the old-            -- socket is not this connection's any more either way.+            -- because the event manager says so.  The reader leaves it for+            -- the sender instead.  Said even if the fork above failed,+            -- since the old socket is not this connection's any more+            -- either way.             atomically $ writeTVar oldRetired True-            fire conn microseconds $ close oldSock

@@ -1,5 +1,3 @@-{-# LANGUAGE ForeignFunctionInterface #-}-{-# LANGUAGE InterruptibleFFI #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE ScopedTypeVariables #-}@@ -9,14 +7,13 @@     migrate, ) where -import Control.Concurrent import Control.Concurrent.Async import Control.Concurrent.STM import qualified Control.Exception as E-import Foreign.C.Types import qualified Network.Socket as NS  import Network.QUIC.Client.Reader+import Network.QUIC.Client.WatchDog import Network.QUIC.Closer import Network.QUIC.Common import Network.QUIC.Config@@ -36,6 +33,10 @@  -- | Running a QUIC client. --   A UDP socket is created according to 'ccServerName' and 'ccPortName'.+--+--   If the server closes the connection with NO_ERROR, an operation waiting+--   on the server throws 'ConnectionIsClosed'.  Otherwise the action runs to+--   its end, and its result is returned. run :: ClientConfig -> (Connection -> IO a) -> IO a -- Don't use handleLogUnit here because of a return value. run conf client = do@@ -53,7 +54,7 @@         Right v -> return v         -- Other exceptions go though.         Left (NextVersion nextVerInfo)-            | verInfo == brokenVersionInfo -> E.throwIO VersionNegotiationFailed+            | nextVerInfo == brokenVersionInfo -> E.throwIO VersionNegotiationFailed             | otherwise -> runClient conf client True nextVerInfo  runClient :: ClientConfig -> (Connection -> IO a) -> Bool -> VersionInfo -> IO a@@ -82,10 +83,16 @@                 -- belongs with them and was missing altogether: it starts at                 -- zero and nothing raised it before the handshake, so 0-RTT                 -- stream data went out against no connection limit at all.-                params <- getPeerParameters conn-                setTxMaxData conn $ initialMaxData params-                setTxMaxStreams conn $ initialMaxStreamsBidi params-                setTxUniMaxStreams conn $ initialMaxStreamsUni params+                --+                -- Only for 0-RTT: without it, these limits are the ones the+                -- handshake gives, and since the setters only raise, starting+                -- from those of a previous connection kept the client to+                -- them when the server had lowered its own since.+                when (remembers conf) $ do+                    params <- getPeerParameters conn+                    setTxMaxData conn $ initialMaxData params+                    setTxMaxStreams conn $ initialMaxStreamsBidi params+                    setTxUniMaxStreams conn $ initialMaxStreamsUni params                 if ccUse0RTT conf                     then wait0RTTReady conn                     else wait1RTTReady conn@@ -103,27 +110,46 @@             c2 = labelMe "concurrently2" >> concurrently_ c1 s3             c3 = labelMe "concurrently3" >> concurrently_ c2 s4             c4 = labelMe "concurrently4" >> concurrently_ c3 s5+            c4' = labelMe "concurrently4'" >> concurrently_ c4 s7+            s7 = labelMe "abort watcher" >> watchAbortRequest conn             c5 =                 labelMe "concurrently5"-                    >> concurrently (c4 `E.catch` \(_ :: InternalControl) -> return ()) s6+                    >> concurrently (c4' `E.catches` handlers) s6+            -- A peer closing with NO_ERROR leaves the application to finish:+            -- what waits on the peer gives up with 'ConnectionIsClosed'.+            handlers =+                [ E.Handler $ \(_ :: InternalControl) -> return ()+                , E.Handler $ \e -> case e of+                    ConnectionIsClosed _ -> return ()+                    _ -> E.throwIO e+                ]             runThreads = snd <$> c5-        when (ccWatchDog conf) $ forkManaged conn "QUIC watchDog" $ watchDog conn-        ex <- E.try runThreads+        when (ccWatchDog conf) $ forkManaged conn "QUIC watchDog" $ watchDog (void $ migrate conn)+        ex <- E.try runThreads >>= abortRequested conn         sendFinal conn         setConnectionClosed conn         closure conn ldcc ex   where     open = createClientConnection conf verInfo-    -- Stopping the readers and seeing them out before the buffers they use-    -- are freed.  This used to free first and stop after, which left every-    -- reader a window in which to use a buffer that had been handed back.+    -- 'killReaders' sees the readers out, which has to happen before the+    -- buffers they use are freed.  This used to free first and stop after,+    -- which left every reader a window in which to use a buffer that had+    -- been handed back.     clse connRes = do         let conn = connResConnection connRes         setDead conn         killReaders conn-        waitManaged conn+        -- Only the closer closes the socket, and there is none when the+        -- peer ended the connection.  No thread is on the socket now.+        owned <- readIORef $ connCloserOwnsSocket conn+        unless owned $ getSocket conn >>= NS.unsafeClose         freeResources conn +-- | Whether this client sends 0-RTT, and so starts from the transport+--   parameters it remembers (RFC 9000 Sec 7.4.1).+remembers :: ClientConfig -> Bool+remembers ClientConfig{..} = ccUse0RTT && is0RTTPossible ccResumption+ createClientConnection :: ClientConfig -> VersionInfo -> IO ConnRes createClientConnection conf@ClientConfig{..} verInfo = do     (sock, peersa) <- clientSocket ccServerName ccPortName@@ -131,7 +157,10 @@     -- nothing releases what it has taken when it throws.  Here that is a     -- socket as well -- 'clse' does not close it, the closer does, and a     -- connection that never began has no closer.-    flip E.onException (NS.close sock) $ do+    --+    -- Nothing below forks, and the reader is only built here, so this+    -- thread is the only one the socket has ever had.+    flip E.onException (NS.unsafeClose sock) $ do         when ccSockConnected $ NS.connect sock peersa         q <- newRecvQ         retire0 <- newTVarIO False@@ -181,7 +210,9 @@                     genSRT             flip E.onException (setDead conn >> freeResources conn) $ do                 setSockConnected conn ccSockConnected-                modifytPeerParameters conn ccResumption+                -- What RFC 9000 Sec 7.4.1 has a client remember, for 0-RTT+                -- alone.+                when (remembers conf) $ modifytPeerParameters conn ccResumption                 let ver = chosenVersion verInfo                 initializeCoder conn InitialLevel $ initialSecrets ver peerCID                 setupCryptoStreams conn -- fixme: cleanup@@ -209,26 +240,3 @@ --   But this can also be used even when the value is 'False'. migrate :: Connection -> IO Bool migrate conn = controlConnection conn ActiveMigration--watchDog :: Connection -> IO ()-watchDog conn = E.bracket c_open_socket c_close_socket loop-  where-    loop s = do-        ret <- c_watch_socket s-        case ret of-            -1 -> loop s-            -2 -> return ()-            _ -> do-                _ <- migrate conn-                -- prevent calling "migrate" frequently-                threadDelay 100000-                loop s--foreign import ccall unsafe "open_socket"-    c_open_socket :: IO CInt--foreign import ccall interruptible "watch_socket"-    c_watch_socket :: CInt -> IO CInt--foreign import ccall unsafe "close_socket"-    c_close_socket :: CInt -> IO CInt

@@ -0,0 +1,53 @@+{-# LANGUAGE ForeignFunctionInterface #-}+{-# LANGUAGE InterruptibleFFI #-}++module Network.QUIC.Client.WatchDog (+    watchDog,+    watchDogWith,+    watchSocket,+) where++import Control.Concurrent+import qualified Control.Exception as E+import Foreign.C.Types++import Network.QUIC.Imports++-- | Running the action on every change of the network the OS reports,+--   until the OS cannot report them.+watchDog :: IO () -> IO ()+watchDog = watchDogWith c_open_socket c_watch_socket c_close_socket++-- | 'watchDog' with the socket calls given.  The watch returns 0 for an+--   event, -1 to be tried again and anything else to give up.  A socket+--   that could not be opened is never watched: a watch on it fails at+--   once, and trying it again spun.+watchDogWith+    :: IO CInt -> (CInt -> IO CInt) -> (CInt -> IO CInt) -> IO () -> IO ()+watchDogWith open watch clse action = E.bracket open release $ \s ->+    when (s >= 0) $ loop s+  where+    release s = when (s >= 0) $ void $ clse s+    loop s = do+        ret <- watch s+        case ret of+            0 -> do+                action+                -- prevent calling "migrate" frequently+                threadDelay 100000+                loop s+            -1 -> loop s+            _ -> return ()++-- | One watch on the socket, as 'watchDog' does.+watchSocket :: CInt -> IO CInt+watchSocket = c_watch_socket++foreign import ccall unsafe "open_socket"+    c_open_socket :: IO CInt++foreign import ccall interruptible "watch_socket"+    c_watch_socket :: CInt -> IO CInt++foreign import ccall unsafe "close_socket"+    c_close_socket :: CInt -> IO CInt

@@ -90,7 +90,11 @@         if isServer conn             then return Nothing             else Just <$> mallocBytes bufsiz `E.onException` free sendbuf-    let freeBufs = free sendbuf >> mapM_ free mrecvbuf+    let freeBufs = do+            free sendbuf+            case mrecvbuf of+                Nothing -> return ()+                Just recvbuf -> free recvbuf     flip E.onException freeBufs $ do         -- This must be called before freeResourcesin runClient.         siz <- encodeCC conn sendbuf bufsiz frame@@ -126,10 +130,14 @@                                         Right (_, sa)                                             | sa /= peersa -> again                                             | otherwise -> return True+                        -- 'killReaders' above has seen the readers out,+                        -- and this runs on the way out of the closer, the+                        -- last thread to send or receive on it: @close(2)@+                        -- is enough.                         clos' = do-                            NS.close sock+                            NS.unsafeClose sock                             -- This is just in case.-                            getSocket conn >>= NS.close+                            getSocket conn >>= NS.unsafeClose                      in (recv', clos')         -- hook         let hook = onCloseCompleted $ connHooks conn@@ -141,6 +149,7 @@         -- of its own misses them.  The closer repeats it for a peer that         -- did not hear it, which is what it is for.         send+        writeIORef (connCloserOwnsSocket conn) True         void $ forkFinally (closer conn pto send recvWithin hook) $ \e -> do             case e of                 Left e' -> connDebugLog conn $ "closure' " <> bhow e'

@@ -7,6 +7,7 @@     module Network.QUIC.Connection.Stream,     module Network.QUIC.Connection.StreamTable,     module Network.QUIC.Connection.Queue,+    module Network.QUIC.Connection.RateControl,     module Network.QUIC.Connection.Role,     module Network.QUIC.Connection.Timeout,     module Network.QUIC.Connection.Types,@@ -17,6 +18,7 @@ import Network.QUIC.Connection.Misc import Network.QUIC.Connection.PacketNumber import Network.QUIC.Connection.Queue+import Network.QUIC.Connection.RateControl import Network.QUIC.Connection.Role import Network.QUIC.Connection.State import Network.QUIC.Connection.Stream

@@ -43,6 +43,9 @@ import Network.QUIC.Imports import Network.QUIC.Parameters import Network.QUIC.Qlog+import Network.QUIC.Recovery+import Network.QUIC.Recovery.Persistent (calcPTO)+import Network.QUIC.Recovery.Types (initialRTT) import Network.QUIC.Types  ----------------------------------------------------------------@@ -346,7 +349,9 @@  ---------------------------------------------------------------- -validatePath :: Connection -> PathInfo -> Maybe CIDInfo -> IO ()+-- | Validating the path, and whether it was validated.  False when no+--   PATH_RESPONSE came in time: the validation is abandoned.+validatePath :: Connection -> PathInfo -> Maybe CIDInfo -> IO Bool validatePath conn pathInfo Nothing = do     pdat <- newPathData     setChallenges conn pathInfo pdat@@ -398,11 +403,20 @@             writeTVar migrationState MigrationStarted             return True -waitResponse :: Connection -> IO ()-waitResponse Connection{..} = atomically $ do-    state <- readTVar migrationState-    check (state == RecvResponse)-    writeTVar migrationState NonMigration+-- | Waiting for the PATH_RESPONSE, for three times the larger of the PTO+--   and the PTO of a new path (RFC 9000 Sec 8.2.4).  It used to wait for+--   ever, and no other migration could start meanwhile.+waitResponse :: Connection -> IO Bool+waitResponse Connection{..} = do+    pto <- getPTO connLDCC+    let Microseconds us = 3 * max pto (calcPTO initialRTT Nothing)+    expired <- registerDelay us+    atomically $ do+        state <- readTVar migrationState+        timedOut <- readTVar expired+        check (state == RecvResponse || timedOut)+        writeTVar migrationState NonMigration+        return (state == RecvResponse)  checkResponse :: Connection -> PathData -> IO () checkResponse Connection{..} pdat = do

@@ -12,8 +12,11 @@     getSocketRetire,     swapSocket,     clearSocket,+    retireSocket,+    closeRetiredSockets,     getPathInfo,     addPathInfo,+    restorePathInfo,     findPathInfo,     getPeerAuthCIDs,     setPeerAuthCIDs,@@ -35,11 +38,21 @@     sendFrames,     closeConnection,     abortConnection,+    watchAbortRequest,+    abortRequested, ) where -import Control.Concurrent.STM (TVar, atomically, writeTVar)+import Control.Concurrent.STM (+    TVar,+    atomically,+    modifyTVar',+    readTVar,+    readTVarIO,+    retry,+    writeTVar,+ ) import qualified Control.Exception as E-import Network.Socket (SockAddr, Socket)+import Network.Socket (SockAddr, Socket, unsafeClose) import qualified System.ThreadManager as T  import Network.QUIC.Connection.Queue@@ -90,6 +103,27 @@ swapSocket Connection{..} sock retire =     atomicModifyIORef' connSocket ((sock, retire),) +-- | Leaving a socket this connection has finished with for the sender to+--   close.  Said by the reader on its way out, so that the only thread+--   which could still be inside a call on it is the sender.+--+-- The close is registered as a resource as well, so that leaving one here+-- is all it takes: a connection whose sender never wakes again closes it+-- in 'freeResources' instead.  Closing an already closed socket does+-- nothing, so it does not matter which gets there first.+retireSocket :: Connection -> Socket -> IO ()+retireSocket conn@Connection{..} sock = do+    atomicModifyIORef' connRetiredSocks $ \socks -> (sock : socks, ())+    addResource conn $ unsafeClose sock++-- | Closing what 'retireSocket' has left.  The caller must not be inside a+--   call on any of them, which is why this is the sender between two sends.+--   Nobody is waiting on them, so @close(2)@ is enough.+closeRetiredSockets :: Connection -> IO ()+closeRetiredSockets Connection{..} = do+    socks <- atomicModifyIORef' connRetiredSocks ([],)+    mapM_ unsafeClose socks+ -- fixme clearSocket :: Connection -> IO Socket clearSocket Connection{..} =@@ -105,6 +139,10 @@   where     add (PeerInfo oldpi _) = PeerInfo newpi $ Just oldpi +-- | Going back to a path, forgetting the one in use.+restorePathInfo :: Connection -> PathInfo -> IO ()+restorePathInfo Connection{..} oldpi = writeIORef peerInfo $ PeerInfo oldpi Nothing+ findPathInfo :: Connection -> SockAddr -> IO (Maybe PathInfo) findPathInfo Connection{..} sa = do     PeerInfo currpi mprevpi <- readIORef peerInfo@@ -222,8 +260,14 @@ waitManaged :: Connection -> IO () waitManaged Connection{..} = T.waitUntilAllGone readers --- | Killing the threads of 'forkManaged'.  One forked after this is---   refused rather than run.+-- | Killing the threads of 'forkManaged' and seeing them out.  One forked+--   after this is refused rather than run.+--+-- 'T.stopAfter' throws and returns: 'throwTo' waits for the exception to be+-- taken, not for the thread to end, so without the wait a caller which goes+-- on to use what those threads were using -- the connection's socket, in+-- 'Network.QUIC.Closer' -- is racing them.  'waitManaged' says why there is+-- no cap on it. killReaders :: Connection -> IO () killReaders conn@Connection{..} = do     -- Telling the reader of the current socket to go.  'stopAfter' below@@ -233,6 +277,7 @@     retired <- getSocketRetire conn     atomically $ writeTVar retired True     T.stopAfter readers (return ()) $ \_ -> return ()+    waitManaged conn  ---------------------------------------------------------------- @@ -259,6 +304,36 @@     quicexc = TransportErrorIsSent err desc  -- | Closing a connection with an application protocol error.+--+-- This asks and returns; the connection's protocol threads act on it+-- ('watchAbortRequest').  It used to throw to the connection's main thread,+-- and a connection that had just ended took the exception where nothing+-- caught it: in its teardown, or on the way out of the thread running it,+-- where the runtime reported it as uncaught.  A server stopping caught one+-- of its connections there now and then.  A connection already over now+-- just never looks.  The first request is the one acted on. abortConnection     :: Connection -> ApplicationProtocolError -> ReasonPhrase -> IO ()-abortConnection conn err desc = E.throwTo (mainThreadId conn) $ Abort err desc+abortConnection Connection{..} err desc =+    atomically $ modifyTVar' connAbortRequest $ \r -> r <|> Just (err, desc)++-- | A connection's ending, as 'abortConnection' asked for it if it did.+--   An application that asks and returns at once can see the protocol+--   threads end before they have acted on it; it is what it asked for all+--   the same.+abortRequested+    :: Connection -> Either E.SomeException a -> IO (Either E.SomeException a)+abortRequested Connection{..} ex@(Right _) = do+    req <- readTVarIO connAbortRequest+    return $ case req of+        Nothing -> ex+        Just (err, desc) -> Left $ E.toException $ Abort err desc+abortRequested _ ex = return ex++-- | Waiting for 'abortConnection' and raising 'Abort' when it is asked:+--   one of a connection's protocol threads, so that it ends the connection+--   as any of them failing does.+watchAbortRequest :: Connection -> IO ()+watchAbortRequest Connection{..} = do+    (err, desc) <- atomically $ readTVar connAbortRequest >>= maybe retry return+    E.throwIO $ Abort err desc

@@ -20,12 +20,11 @@     inc pn = (pn + 1, pn)  ------------------------------------------------------------------- Peer's max packet number for RTT1+-- Peer's largest packet number, which a truncated one is decoded against -getPeerPacketNumber :: Connection -> IO PacketNumber-getPeerPacketNumber Connection{..} = readIORef peerPacketNumber+getPeerPacketNumber :: Connection -> EncryptionLevel -> IO PacketNumber+getPeerPacketNumber Connection{..} lvl = readIORef (peerPacketNumber ! lvl) -setPeerPacketNumber :: Connection -> PacketNumber -> IO ()-setPeerPacketNumber Connection{..} n = atomicModifyIORef'' peerPacketNumber set-  where-    set m = max m n+setPeerPacketNumber :: Connection -> EncryptionLevel -> PacketNumber -> IO ()+setPeerPacketNumber Connection{..} lvl n =+    atomicModifyIORef'' (peerPacketNumber ! lvl) $ max n

@@ -1,7 +1,6 @@ module Network.QUIC.Connection.Queue where  import Control.Concurrent.STM-import Network.Control (getRate)  import Network.QUIC.Connection.Types import Network.QUIC.Stream@@ -10,7 +9,7 @@ ----------------------------------------------------------------  takeInput :: Connection -> IO Input-takeInput conn = atomically $ readTQueue (inputQ conn)+takeInput conn = atomically $ readTQueue (inputQ conn) `orElse` peerClosedSTM conn  putInput :: Connection -> Input -> IO () putInput conn inp = atomically $ writeTQueue (inputQ conn) inp@@ -29,17 +28,32 @@ ----------------------------------------------------------------  takeOutputSTM :: Connection -> STM Output-takeOutputSTM conn = readTQueue (outputQ conn)+takeOutputSTM conn = do+    out <- readTQueue (outputQ conn)+    modifyTVar' (queuedControls conn) (subtract $ countControls out)+    return out  tryTakeOutput :: Connection -> IO (Maybe Output)-tryTakeOutput conn = atomically $ tryReadTQueue (outputQ conn)+tryTakeOutput conn = atomically $ do+    mout <- tryReadTQueue (outputQ conn)+    case mout of+        Nothing -> return ()+        Just out -> modifyTVar' (queuedControls conn) (subtract $ countControls out)+    return mout  tryPeekOutput :: Connection -> IO (Maybe Output) tryPeekOutput conn = atomically $ tryPeekTQueue (outputQ conn)  putOutput :: Connection -> Output -> IO ()-putOutput conn out = atomically $ writeTQueue (outputQ conn) out+putOutput conn out = atomically $ do+    writeTQueue (outputQ conn) out+    modifyTVar' (queuedControls conn) (+ countControls out) +-- | Control frames waiting for the sender, which a peer that keeps the+--   congestion window shut can make us pile up (CVE-2024-22189).+getQueuedControls :: Connection -> IO Int+getQueuedControls conn = readTVarIO $ queuedControls conn+ isEmptyOutputSTM :: Connection -> STM Bool isEmptyOutputSTM conn = isEmptyTQueue $ outputQ conn @@ -56,6 +70,10 @@     outs <- flushTQueue (outputQ conn)     let (found, rest) = pick outs     mapM_ (writeTQueue (outputQ conn)) rest+    case found of+        Nothing -> return ()+        Just ppkt ->+            modifyTVar' (queuedControls conn) (subtract $ countControls $ OutRetrans ppkt)     return found   where     pick [] = (Nothing, [])@@ -85,13 +103,3 @@  isEmptyStreamSTM :: Connection -> STM Bool isEmptyStreamSTM conn = isEmptyTQueue $ sharedSendStreamQ $ shared conn--------------------------------------------------------------------outputLimit :: Int-outputLimit = 10--rateOK :: Connection -> IO Bool-rateOK conn = do-    rate <- getRate $ outputRate conn-    return $ rate < outputLimit

@@ -0,0 +1,132 @@+{-# LANGUAGE OverloadedStrings #-}++-- | Limits against a peer which makes us answer frames.+module Network.QUIC.Connection.RateControl (+    -- * Limits+    PerSecond,+    pathAndCIDFrameLimit,+    pathResponseLimit,+    queuedControlLimit,+    statelessResetLimit,+    statelessResetSize,+    statelessResetSizeFor,++    -- * Enforcing+    enforcePathAndCIDFrameLimit,+    withinPathResponseLimit,+    enforceQueuedControlLimit,+    withinStatelessResetLimit,+) where++import Network.Control (Rate, addRate)++import Network.QUIC.Connection.Misc+import Network.QUIC.Connection.Queue+import Network.QUIC.Connection.Types+import Network.QUIC.Imports+import Network.QUIC.Types++----------------------------------------------------------------++-- | How many events a counting window may hold.  A window starts at its+--   first event and lasts about a second, so a burst at the end of one+--   and the start of the next can reach twice the limit.+--+-- A peer can make us answer frames and keep the answers from leaving by+-- not opening its congestion window (CVE-2023-49295, CVE-2024-22189).+-- The limits are against that.  They are fixed, not configurable.  An+-- @enforce@ limit closes the connection with 'InternalError' when it is+-- exceeded; a @within@ limit drops what is past it, which is done only+-- for what the peer retransmits by itself.+--+-- Frames which must be answered and which another limit already bounds+-- are not rate-controlled.  STOP_SENDING and RESET_STREAM are bounded by+-- MAX_STREAMS, and a peer cancelling many streams at once does nothing+-- wrong.  A STOP_SENDING is answered once per stream.  NEW_CONNECTION_ID+-- is always processed, as dropping it would lose its Retire Prior To.+newtype PerSecond = PerSecond Int deriving (Eq, Ord, Show)++-- | Per connection, PATH_CHALLENGE, PATH_RESPONSE, NEW_CONNECTION_ID and+--   RETIRE_CONNECTION_ID received count together.  Enforced.+pathAndCIDFrameLimit :: PerSecond+pathAndCIDFrameLimit = PerSecond 32++-- | Per connection, PATH_CHALLENGE frames answered.  The rest are+--   dropped: the peer sends a new one if it gets no PATH_RESPONSE.+pathResponseLimit :: PerSecond+pathResponseLimit = PerSecond 10++-- | Per connection, control frames waiting for the sender, those of lost+--   packets to resend included.  Data, which flow control bounds, is not+--   counted.  Enforced.  With what the congestion window lets be in+--   flight, this bounds the RETIRE_CONNECTION_ID frames a NEW_CONNECTION_ID+--   flood asks for.  The value is quic-go's.+queuedControlLimit :: Int+queuedControlLimit = 16384++-- | Per server, stateless resets sent.  The rest are not sent.+statelessResetLimit :: PerSecond+statelessResetLimit = PerSecond 20++-- | The largest stateless reset we send, in octets.+statelessResetSize :: Int+statelessResetSize = 1280++-- | The size of a stateless reset in answer to a packet of this many+--   octets, if one can be sent.+--+-- Always smaller than the packet (RFC 9000, section 10.3.3): a reset looks+-- like a packet for an unknown connection, so two endpoints could answer+-- each other's resets for ever.  Half of it, so that such a loop ends in a+-- few rounds, and one octet less for a packet of 43 octets or fewer+-- (section 10.3).  None below the 21 octets a reset needs.  Being smaller,+-- it is also under three times the packet (section 10.3).+statelessResetSizeFor :: Int -> Maybe Int+statelessResetSizeFor bytes+    | siz < 21 = Nothing+    | otherwise = Just siz+  where+    siz+        | bytes <= 43 = bytes - 1+        | otherwise = min statelessResetSize (bytes `div` 2)++----------------------------------------------------------------++-- Adds n to the window and tells whether it is still within the limit.+-- The count includes what is being added, so the limit itself is let+-- through with '<='.  '<' let one fewer through than the name says.+addWithin :: Rate -> Int -> PerSecond -> IO Bool+addWithin rate n (PerSecond lim) = (<= lim) <$> addRate rate n++-- | Counting the path and CID frames of a packet against+--   'pathAndCIDFrameLimit'.+enforcePathAndCIDFrameLimit :: Connection -> [Frame] -> IO ()+enforcePathAndCIDFrameLimit conn frames = do+    let n = length $ filter isPathOrCID frames+    when (n /= 0) $ do+        ok <- addWithin (pathAndCIDFrameRate conn) n pathAndCIDFrameLimit+        unless ok $+            closeConnection conn InternalError "Too many path or CID frames"+  where+    isPathOrCID PathChallenge{} = True+    isPathOrCID PathResponse{} = True+    isPathOrCID NewConnectionID{} = True+    isPathOrCID RetireConnectionID{} = True+    isPathOrCID _ = False++-- | Whether a PATH_CHALLENGE may be answered under 'pathResponseLimit'.+withinPathResponseLimit :: Connection -> IO Bool+withinPathResponseLimit conn =+    addWithin (pathResponseRate conn) 1 pathResponseLimit++-- | Closing the connection if more control frames than+--   'queuedControlLimit' wait for the sender.+enforceQueuedControlLimit :: Connection -> IO ()+enforceQueuedControlLimit conn = do+    queued <- getQueuedControls conn+    when (queued > queuedControlLimit) $+        closeConnection conn InternalError "Too many control frames queued"++-- | Whether a stateless reset may be sent under 'statelessResetLimit'.+withinStatelessResetLimit :: Rate -> IO Bool+withinStatelessResetLimit rate = addWithin rate 1 statelessResetLimit

@@ -78,22 +78,28 @@  -- | Waiting until 0-RTT data can be sent. wait0RTTReady :: Connection -> IO ()-wait0RTTReady Connection{..} = atomically $ do-    cs <- readTVar $ connectionState connState-    check (cs >= ReadyFor0RTT && cs /= Closed)+wait0RTTReady conn@Connection{..} = atomically $ ready `orElse` peerClosedSTM conn+  where+    ready = do+        cs <- readTVar $ connectionState connState+        check (cs >= ReadyFor0RTT && cs /= Closed)  -- | Waiting until 1-RTT data can be sent. wait1RTTReady :: Connection -> IO ()-wait1RTTReady Connection{..} = atomically $ do-    cs <- readTVar $ connectionState connState-    check (cs >= ReadyFor1RTT && cs /= Closed)+wait1RTTReady conn@Connection{..} = atomically $ ready `orElse` peerClosedSTM conn+  where+    ready = do+        cs <- readTVar $ connectionState connState+        check (cs >= ReadyFor1RTT && cs /= Closed)  -- | For clients, waiting until HANDSHAKE_DONE is received. --   For servers, waiting until a TLS stack reports that the handshake is complete. waitEstablished :: Connection -> IO ()-waitEstablished Connection{..} = atomically $ do-    cs <- readTVar $ connectionState connState-    check (cs >= Established && cs /= Closed)+waitEstablished conn@Connection{..} = atomically $ ready `orElse` peerClosedSTM conn+  where+    ready = do+        cs <- readTVar $ connectionState connState+        check (cs >= Established && cs /= Closed)  ---------------------------------------------------------------- 

@@ -11,12 +11,14 @@     checkRxMaxStreams,     updatePeerStreamId,     checkStreamIdRoom,+    releasePeerStream, ) where  import Control.Concurrent.STM import qualified Control.Exception as E  import Network.QUIC.Connection.Misc+import Network.QUIC.Connection.Timeout import Network.QUIC.Connection.Types import Network.QUIC.Connector import Network.QUIC.Imports@@ -35,13 +37,14 @@     return (base - (currentStream !>>. 2))  waitMyNewStreamId :: Connection -> IO StreamId-waitMyNewStreamId Connection{..} = get myStreamId+waitMyNewStreamId conn@Connection{..} = get conn myStreamId  waitMyNewUniStreamId :: Connection -> IO StreamId-waitMyNewUniStreamId Connection{..} = get myUniStreamId+waitMyNewUniStreamId conn@Connection{..} = get conn myUniStreamId -get :: TVar Concurrency -> IO Int-get tvar = atomically $ do+get :: Connection -> TVar Concurrency -> IO Int+get conn tvar = atomically $ do+    checkPeerClosedSTM conn     conc@Concurrency{..} <- readTVar tvar     let streamType = currentStream .&. 0b11         StreamIdBase base = maxStreams@@ -145,3 +148,24 @@          in if base' - base >= max 1 (initialStreams `div` 2)                 then (conc'{maxStreams = StreamIdBase base'}, Just base')                 else (conc', Nothing)++-- | Counting a stream of the peer's as done with, and announcing a new+--   MAX_STREAMS if that is due.  Nothing for a stream we opened.+releasePeerStream :: Connection -> StreamId -> IO ()+releasePeerStream conn sid+    | peers = do+        mx <- checkStreamIdRoom conn dir+        case mx of+            Nothing -> return ()+            Just nms -> do+                sendFrames conn RTT1Level [MaxStreams dir nms]+                fire conn (Microseconds 50000) $+                    sendFrames conn RTT1Level [MaxStreams dir nms]+    | otherwise = return ()+  where+    peers+        | isClient conn = isServerInitiated sid+        | otherwise = isClientInitiated sid+    dir+        | isUnidirectional sid = Unidirectional+        | otherwise = Bidirectional

@@ -1,3 +1,4 @@+{-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-}  module Network.QUIC.Connection.StreamTable (@@ -6,14 +7,16 @@     findStream,     addStream,     delStream,-    keepDepartedStream,-    noteDepartedRxFrame,+    departStream,+    Arrival (..),+    noteRxArrival,     initialRxMaxStreamData,     setupCryptoStreams,     clearCryptoStream,     getCryptoStream, ) where +import Control.Concurrent.STM import qualified Data.IntMap.Strict as IntMap import qualified Data.IntSet as IntSet @@ -53,7 +56,7 @@         | otherwise = (os, False)  findStream :: Connection -> StreamId -> IO (Maybe Stream)-findStream Connection{..} sid = lookupStream sid <$> readIORef streamTable+findStream Connection{..} sid = lookupStream sid <$> readTVarIO streamTable  addStream :: Connection -> StreamId -> IO Stream addStream conn@Connection{..} sid = do@@ -71,74 +74,103 @@                     serverParams = getMyParameters conn                     rxLim = serverInitial sid serverParams                 newStream conn sid txLim rxLim-    atomicModifyIORef'' streamTable $ insertStream sid strm+    atomically $ modifyTVar' streamTable $ insertStream sid strm     return strm  delStream :: Connection -> Stream -> IO () delStream Connection{..} strm =-    atomicModifyIORef'' streamTable $ deleteStream $ streamId strm+    atomically $ modifyTVar' streamTable $ deleteStream $ streamId strm  ---------------------------------------------------------------- --- | Keeping account of a stream the application has closed before the peer---   finished it.+-- | Taking a stream the application is done with out of the table, and+--   answering with what its window is owed back and whether anything is+--   kept of it. ----- A frame that arrives for a stream no longer in the table is dropped, and--- what the peer spent on it is then counted by nobody: not as received,--- since we never looked at it, and not as consumed, since nobody will read--- it.  The window we advertise falls that much behind what the peer--- believes it has spent, for the rest of the connection, and a server that--- answers requests without reading their bodies runs its peers out of--- window.  'Network.QUIC.IO.releaseStream' gives back what had arrived by--- the time of the close; this is for what arrives after it.+-- A frame that arrives for a stream no longer in the table is dropped, but+-- the peer spent its window on it all the same.  What has been seen by now+-- is owed back here; what arrives after is owed by what is kept, which also+-- holds it to the limits it was sent under.  Nothing is kept for a stream+-- the peer has finished, its final size known and accounted for, nor for+-- one we only send on. ----- Nothing is kept for a stream the peer has finished: its final size is--- known and accounted for, and anything that still arrives for it lies--- inside that and has been counted already.-keepDepartedStream :: Connection -> Stream -> IO ()-keepDepartedStream Connection{..} strm = do-    b <- getRxBounds strm-    unless (settled b) $-        atomicModifyIORef'' departedStreams $-            IntMap.insert (streamId strm) b+-- One transaction, and 'noteRxArrival' another, so that a frame is seen+-- either before the stream leaves the table, and counted in what is owed+-- here, or after, and counted in what is kept.  Done in steps, a frame+-- arriving in between was counted in neither, and a FIN lost that way left+-- the stream kept for good.+departStream :: Connection -> Stream -> IO (Int, Bool)+departStream conn strm = do+    -- The application's own, which only it moves.+    consumed <- getRxConsumed strm+    lim <- getRxMaxStreamData strm+    atomically $ do+        modifyTVar' (streamTable conn) $ deleteStream sid+        owed <- takeRxOwedSTM strm consumed+        RxBounds{..} <- getRxBoundsSTM strm+        let kept = not sendOnly && isNothing rxFinal+        when kept $+            modifyTVar' (departedStreams conn) $+                IntMap.insert sid $+                    Departed rxHighest lim+        return (owed, kept)+  where+    sid = streamId strm+    sendOnly+        | isClient conn = isClientInitiatedUnidirectional sid+        | otherwise = isServerInitiatedUnidirectional sid --- | Is there nothing more this stream can owe?-settled :: RxBounds -> Bool-settled RxBounds{..} = rxFinal == Just rxCounted+-- | Where a STREAM or RESET_STREAM frame lands, with what it counts there.+data Arrival+    = -- | A stream in the table, with how far the frame takes it+      ArrivedLive Stream (Either FinalSizeProblem Int)+    | -- | A stream the application has closed and the peer has not finished:+      --   how far the frame takes it and whether it has ended now, or the+      --   transport error the stream would have refused it with+      ArrivedDeparted (Either (TransportError, ReasonPhrase) (Int, Bool))+    | -- | Neither: a stream to open, or one closed and done with+      ArrivedNowhere --- | Noting a frame that arrived for a stream the application has closed,---   and answering with what the connection's window is owed for it.+-- | Finding the stream a frame reaching @end@ is for and noting the frame+--   there, as one transaction against 'departStream'. ----- The peer's own flow control counts the offsets it has sent, not the--- octets that reached us, so what is owed is measured the same way: the--- furthest point of the stream anything has reached, less what has been--- accounted for already.  A retransmission does not move that point and so--- is owed nothing, which is what keeps this from counting twice -- the--- reassembly that tells a copy from new data went with the stream.-noteDepartedRxFrame :: Connection -> StreamId -> Int -> Bool -> IO Int-noteDepartedRxFrame Connection{..} sid end fin =-    atomicModifyIORef' departedStreams note+-- For a stream the application has closed, as for a live one, what moves is+-- how far the frame takes it past what has been seen, which the connection+-- counts as received and, nobody being left to read it, as consumed too.+-- A copy of what was seen moves nothing.  @unread@ says nobody will read+-- what moves on a live stream either, the tail a RESET_STREAM gives up: it+-- is counted as credited here, or 'departStream' could credit it again.+noteRxArrival :: Connection -> StreamId -> Int -> Bool -> Bool -> IO Arrival+noteRxArrival Connection{..} sid end fin unread = atomically $ do+    found <- lookupStream sid <$> readTVar streamTable+    case found of+        Just strm -> do+            r <- noteRxFrameSTM strm end fin+            case r of+                Right n | unread -> addRxCreditedSTM strm n+                _ -> return ()+            return $ ArrivedLive strm r+        Nothing -> do+            tbl <- readTVar departedStreams+            case IntMap.lookup sid tbl of+                Nothing -> return ArrivedNowhere+                Just d -> do+                    let (tbl', r) = note tbl d+                    writeTVar departedStreams tbl'+                    return $ ArrivedDeparted r   where-    note tbl = case IntMap.lookup sid tbl of-        Nothing -> (tbl, 0)-        Just b ->-            let b' = account b-             in ( if settled b' then IntMap.delete sid tbl else IntMap.insert sid b' tbl-                , rxCounted b' - rxCounted b-                )-    account b@RxBounds{..} =-        b-            { rxCounted = reached-            , rxHighest = max rxHighest end-            , rxFinal = if fin then Just end else rxFinal-            }-      where-        -- Never past the end the peer has said the stream has: a frame-        -- claiming to reach beyond it is for the live path to refuse, and-        -- this one is not the place to answer it.-        reached = case (if fin then Just end else rxFinal) of-            Just f -> min f $ max rxCounted end-            Nothing -> max rxCounted end+    note tbl d@Departed{..}+        -- FLOW CONTROL: MAX_STREAM_DATA: recv: rejecting if over my limit+        | end > depLimit =+            (tbl, Left (FlowControlError, "Flow control error for a closed stream"))+        -- RFC 9000 Sec 4.5: "even after a stream is closed".+        | fin && end < depHighest =+            (tbl, Left (FinalSizeError, "a final size below what the stream has reached"))+        | fin = (IntMap.delete sid tbl, Right (end - depHighest, True))+        | otherwise =+            ( IntMap.insert sid d{depHighest = max depHighest end} tbl+            , Right (max 0 (end - depHighest), False)+            )  initialRxMaxStreamData :: Connection -> StreamId -> Int initialRxMaxStreamData conn sid@@ -165,16 +197,16 @@  setupCryptoStreams :: Connection -> IO () setupCryptoStreams conn@Connection{..} = do-    stbl0 <- readIORef streamTable+    stbl0 <- readTVarIO streamTable     stbl <- insertCryptoStreams conn stbl0-    writeIORef streamTable stbl+    atomically $ writeTVar streamTable stbl  clearCryptoStream :: Connection -> EncryptionLevel -> IO () clearCryptoStream Connection{..} lvl =-    atomicModifyIORef'' streamTable $ deleteCryptoStream lvl+    atomically $ modifyTVar' streamTable $ deleteCryptoStream lvl  ----------------------------------------------------------------  getCryptoStream :: Connection -> EncryptionLevel -> IO (Maybe Stream) getCryptoStream Connection{..} lvl =-    lookupCryptoStream lvl <$> readIORef streamTable+    lookupCryptoStream lvl <$> readTVarIO streamTable

@@ -266,10 +266,15 @@     -- ^ The socket and the variable which retires its reader: setting it     --   tells the reader that the socket is not its to read any more.     --   One reference, so that the two cannot be swapped apart.+    , connRetiredSocks  :: IORef [Socket]+    -- ^ Sockets a reader has left and nobody is in any more.  The sender+    --   closes them, because it is the one thread which could otherwise+    --   be inside a send on one, and the connection's teardown closes+    --   what is left.     , genStatelessResetToken :: CID -> StatelessResetToken     , readers           :: T.ThreadManager     , mainThreadId      :: ThreadId-    , controlRate       :: Rate+    , pathAndCIDFrameRate :: Rate     , -- Info       roleInfo          :: IORef RoleInfo     , quicVersionInfo   :: IORef VersionInfo@@ -285,17 +290,21 @@       inputQ            :: InputQ     , cryptoQ           :: CryptoQ     , outputQ           :: OutputQ-    , outputRate        :: Rate+    , pathResponseRate  :: Rate+    , queuedControls    :: TVar Int+    -- ^ Control frames in 'outputQ' which the sender has not taken yet.     , shared            :: Shared     , delayedAckCount   :: IORef Int     , delayedAckCancel  :: IORef (IO ())     , -- State-      peerPacketNumber  :: IORef PacketNumber -- for RTT1-    , streamTable       :: IORef StreamTable+      peerPacketNumber  :: Array EncryptionLevel (IORef PacketNumber)+    -- ^ The largest the peer has sent, by level.  0-RTT and 1-RTT share+    --   one, as they share a packet number space.+    , streamTable       :: TVar StreamTable     , -- | What a stream the application has closed still owes the       -- connection's window, for as long as the peer has not finished it.-      -- See 'Network.QUIC.Connection.StreamTable.keepDepartedStream'.-      departedStreams   :: IORef (IntMap RxBounds)+      -- See 'Network.QUIC.Connection.StreamTable.departStream'.+      departedStreams   :: TVar (IntMap Departed)     , myStreamId        :: TVar Concurrency -- C:0 S:1     , myUniStreamId     :: TVar Concurrency -- C:2 S:3     , peerStreamId      :: IORef Concurrency -- C:1 S:0@@ -310,6 +319,9 @@     , bytesTx           :: IORef Int     , bytesRx           :: IORef Int     , connDone          :: TVar Bool+    , connAbortRequest  :: TVar (Maybe (ApplicationProtocolError, ReasonPhrase))+    -- ^ What 'Network.QUIC.Connection.Misc.abortConnection' asked to close+    --   the connection with, for its protocol threads to act on.     , -- TLS       pendingQ          :: Array EncryptionLevel (TVar [ReceivedPacket])     , ciphers           :: IOArray EncryptionLevel Cipher@@ -330,6 +342,8 @@     , connResources     :: IORef (IO ())     , connCloseSent     :: IORef Bool     -- ^ Whether the peer has been told the connection is over+    , connCloserOwnsSocket :: IORef Bool+    -- ^ Whether the closer has taken the client's socket to close     , connAuthFailures  :: IORef Int     -- ^ Packets that have failed authentication, over the whole connection     , connKeyPackets    :: IORef Int@@ -385,7 +399,7 @@     -- Manage     readers           <- T.newThreadManager     mainThreadId      <- myThreadId-    controlRate       <- newRate+    pathAndCIDFrameRate <- newRate     -- Info     roleInfo          <- newIORef roleinfo     quicVersionInfo   <- newIORef origVersionInfo@@ -398,14 +412,15 @@     inputQ            <- newTQueueIO     cryptoQ           <- newTQueueIO     outputQ           <- newTQueueIO-    outputRate        <- newRate+    pathResponseRate  <- newRate+    queuedControls    <- newTVarIO 0     shared            <- newShared     delayedAckCount   <- newIORef 0     delayedAckCancel  <- newIORef (return ())     -- State-    peerPacketNumber  <- newIORef 0-    streamTable       <- newIORef emptyStreamTable-    departedStreams   <- newIORef IntMap.empty+    peerPacketNumber  <- newPeerPacketNumber+    streamTable       <- newTVarIO emptyStreamTable+    departedStreams   <- newTVarIO IntMap.empty     myStreamId        <- newTVarIO (newConcurrency rl Bidirectional 0)     myUniStreamId     <- newTVarIO (newConcurrency rl Unidirectional 0)     peerStreamId      <- newIORef peerConcurrency@@ -420,6 +435,8 @@     bytesTx           <- newIORef 0     bytesRx           <- newIORef 0     connDone          <- newTVarIO False+    connAbortRequest  <- newTVarIO Nothing+    connRetiredSocks  <- newIORef []     -- TLS     pendingQ          <- makePendingQ     ciphers           <- newArray (InitialLevel, RTT1Level) defaultCipher@@ -438,10 +455,15 @@     decryptBuf        <- mallocBytes bufsiz -- used receiver     connResources     <- newIORef (free encodeBuf >> free encryptBuf >> free decryptBuf)     connCloseSent     <- newIORef False+    connCloserOwnsSocket <- newIORef False     connAuthFailures  <- newIORef 0     connKeyPackets    <- newIORef 0     -- Recovery-    let put x = atomically $ writeTQueue outputQ $ OutRetrans x+    -- Counted as 'putOutput' does, which this cannot use.+    let put x = atomically $ do+            let out = OutRetrans x+            writeTQueue outputQ out+            modifyTVar' queuedControls (+ countControls out)     connLDCC          <- newLDCC connState connQLog put     return Connection{..}   where@@ -514,6 +536,18 @@  ---------------------------------------------------------------- +newPeerPacketNumber :: IO (Array EncryptionLevel (IORef PacketNumber))+newPeerPacketNumber = do+    ini <- newIORef 0+    app <- newIORef 0+    hnd <- newIORef 0+    return $+        listArray+            (InitialLevel, RTT1Level)+            [ini, app, hnd, app] -- Initial, 0-RTT, Handshake, 1-RTT++----------------------------------------------------------------+ newtype Input = InpStream Stream deriving (Show) data Crypto = InpHandshake EncryptionLevel ByteString deriving (Show) @@ -522,6 +556,22 @@     | OutHandshake [(EncryptionLevel, ByteString)]     | OutRetrans PlainPacket +-- | The control frames of an output, which 'queuedControls' counts.+--   Not data, which flow control bounds, nor what is never resent.+countControls :: Output -> Int+countControls (OutControl _ frames) = length $ filter isControl frames+countControls (OutRetrans (PlainPacket _ plain)) = length $ filter isControl $ plainFrames plain+countControls OutHandshake{} = 0++isControl :: Frame -> Bool+isControl Padding{} = False+isControl Ack{} = False+isControl StreamF{} = False+isControl CryptoF{} = False+isControl Datagram{} = False+isControl PathResponse{} = False+isControl _ = True+ type InputQ = TQueue Input type CryptoQ = TQueue Crypto type OutputQ = TQueue Output@@ -534,7 +584,8 @@  data Shared = Shared     { sharedCloseSent :: IORef Bool-    , sharedCloseReceived :: IORef Bool+    , sharedCloseReceived :: TVar (Maybe ReasonPhrase)+    -- ^ The reason of a CONNECTION_CLOSE(NO_ERROR) from the peer     , shared1RTTReady :: IORef Bool     , sharedSendStreamQ :: SendStreamQ     }@@ -542,7 +593,7 @@ newShared :: IO Shared newShared = do     sharedCloseSent <- newIORef False-    sharedCloseReceived <- newIORef False+    sharedCloseReceived <- newTVarIO Nothing     shared1RTTReady <- newIORef False     sharedSendStreamQ <- newTQueueIO     return Shared{..}@@ -551,3 +602,26 @@  mainDone :: Connection -> IO () mainDone conn = atomically $ writeTVar (connDone conn) True++-- | The peer has closed the connection: we drain, so nothing is sent back+--   (RFC 9000 Sec 10.2.2), and what is waiting on the peer gives up.+setPeerClosed :: Connection -> ReasonPhrase -> IO ()+setPeerClosed conn reason = do+    writeIORef (connCloseSent conn) True+    atomically $ writeTVar (sharedCloseReceived $ shared conn) $ Just reason++isPeerClosed :: Connection -> IO Bool+isPeerClosed conn = isJust <$> readTVarIO (sharedCloseReceived $ shared conn)++-- | Waiting for the peer to close, then throwing 'ConnectionIsClosed'.+--   An alternative for a wait on the peer.+peerClosedSTM :: Connection -> STM a+peerClosedSTM conn = do+    mr <- readTVar $ sharedCloseReceived $ shared conn+    case mr of+        Nothing -> retry+        Just reason -> throwSTM $ ConnectionIsClosed reason++-- | Throwing 'ConnectionIsClosed' if the peer has closed.+checkPeerClosedSTM :: Connection -> STM ()+checkPeerClosedSTM conn = peerClosedSTM conn `orElse` return ()

@@ -1,3 +1,7 @@ module Network.QUIC.Connection.Types where +import Control.Concurrent.STM (STM)+ data Connection++peerClosedSTM :: Connection -> STM a

@@ -243,7 +243,7 @@ ----------------------------------------------------------------  setPeerParams :: Connection -> TLS.Context -> [ExtensionRaw] -> IO ()-setPeerParams conn _ctx peerExts = do+setPeerParams conn ctx peerExts = do     tpId <- extensionIDForTtransportParameter <$> getVersion conn     case getTP tpId peerExts of         Nothing ->@@ -257,6 +257,10 @@             checkAuthCIDs params             checkInvalid params             setParams params+            -- RFC 9000 Sec 18.2: "UDP datagrams with payloads larger than+            -- this limit are not likely to be processed by the receiver."+            siz <- getMaxPacketSize conn+            setMaxPacketSize conn $ min siz $ maxUdpPayloadSize params             qlogParamsSet conn (params, "remote")             if isClient conn                 then@@ -290,8 +294,10 @@         -- exactly that many from the end of a datagram to compare against.         -- A token of some other length could never match one, so it is not         -- a token.-        forM_ (statelessResetToken params) $ \(StatelessResetToken srt) ->-            when (Short.length srt /= 16) sendCCParamError+        case statelessResetToken params of+            Nothing -> return ()+            Just (StatelessResetToken srt) ->+                when (Short.length srt /= 16) sendCCParamError         when (isServer conn) $ do             when (isJust $ originalDestinationConnectionId params) sendCCParamError             when (isJust $ preferredAddress params) sendCCParamError@@ -323,9 +329,20 @@                     newMaxDatagram = maxDatagramFrameSize params                 when (storedMaxDatagram > 0 && newMaxDatagram < storedMaxDatagram) $                     E.throwIO WrongDatagramSizeParameter+                -- RFC 9000 Sec 7.4.1: "If 0-RTT data is accepted by the+                -- server, the server MUST NOT reduce any limits".  The limits+                -- only ever rise here, so one lowered all the same was kept+                -- at the old value, and the server closed the connection over+                -- what it had itself allowed before.+                minfo <- TLS.contextGetInformation ctx+                let accepted = (minfo >>= TLS.infoTLS13HandshakeMode) == Just TLS.RTT0+                when (accepted && lowered storedParams params) $+                    E.throwIO Wrong0RTTLimitParameter          setPeerParameters conn params-        mapM_ (setPeerStatelessResetToken conn) $ statelessResetToken params+        case statelessResetToken params of+            Nothing -> return ()+            Just srt -> setPeerStatelessResetToken conn srt         setTxMaxData conn $ initialMaxData params         -- My 'maxIdleTimeout' is already set 'minIdleTimeout'         -- This selects the minimum of mine and peer's.@@ -334,6 +351,20 @@         setTxMaxStreams conn $ initialMaxStreamsBidi params         setTxUniMaxStreams conn $ initialMaxStreamsUni params +    lowered old new =+        or+            [ f new < f old+            | f <-+                [ activeConnectionIdLimit+                , initialMaxData+                , initialMaxStreamDataBidiLocal+                , initialMaxStreamDataBidiRemote+                , initialMaxStreamDataUni+                , initialMaxStreamsBidi+                , initialMaxStreamsUni+                ]+            ]+     serverVersionNegotiation Nothing = return ()     serverVersionNegotiation (Just peerVerInfo) = do         myVerInfo <- getVersionInfo conn@@ -373,6 +404,7 @@ sendCCTLSError conn (TLS.HandshakeFailed (TLS.Error_Misc "WrongTransportParameter")) = closeConnection conn TransportParameterError "Transport parameter error" sendCCTLSError conn (TLS.HandshakeFailed (TLS.Error_Misc "WrongVersionInformation")) = closeConnection conn VersionNegotiationError "Version negotiation error" sendCCTLSError conn (TLS.HandshakeFailed (TLS.Error_Misc "WrongDatagramSizeParameter")) = closeConnection conn ProtocolViolation "0-RTT datagram size violation"+sendCCTLSError conn (TLS.HandshakeFailed (TLS.Error_Misc "Wrong0RTTLimitParameter")) = closeConnection conn ProtocolViolation "0-RTT limits lowered" sendCCTLSError conn e = closeConnection conn err msg   where     tlserr = getErrorCause e

@@ -68,13 +68,15 @@         eblocked <- checkBlocked s len wait         case eblocked of             Right n+                -- Counted before queued, so that a reset in between takes+                -- it into the final size.                 | len == n -> do-                    putSendStreamQ conn $ TxStreamData s dats len False                     addTx conn s n+                    putSendStreamQ conn $ TxStreamData s dats len False                 | otherwise -> do                     let (dats1, dats2) = split n dats-                    putSendStreamQ conn $ TxStreamData s dats1 n False                     addTx conn s n+                    putSendStreamQ conn $ TxStreamData s dats1 n False                     flowControl dats2 (len - n) False             Left blocked -> do                 -- Read each time round rather than once: the handshake may@@ -112,6 +114,7 @@ checkBlocked :: Stream -> Int -> Bool -> IO (Either Blocked Int) checkBlocked s len wait = atomically $ do     let conn = streamConnection s+    checkPeerClosedSTM conn     strmFlow <- readStreamFlowTx s     connFlow <- readConnectionFlowTx conn     let strmWindow = txWindowSize strmFlow@@ -144,8 +147,16 @@  -- | Closing a stream without an error. --   This sends FIN if necessary.+--   If the peer has not finished sending on it, this also sends+--   STOP_SENDING with error code 0; see 'closeStreamWith'. closeStream :: Stream -> IO ()-closeStream s = do+closeStream s = closeStreamWith s $ ApplicationProtocolError 0++-- | 'closeStream' with the error code of the STOP_SENDING sent if the+--   peer has not finished sending on the stream.  HTTP\/3, for one, wants+--   H3_NO_ERROR there (RFC 9114, section 4.1).+closeStreamWith :: Stream -> ApplicationProtocolError -> IO ()+closeStreamWith s aerr = do     let conn = streamConnection s         sid = streamId s         -- A unidirectional stream of the peer's has no sending side here.@@ -154,76 +165,69 @@         receiveOnly =             (isClient conn && isServerInitiatedUnidirectional sid)                 || (isServer conn && isClientInitiatedUnidirectional sid)-    closed <- isConnectionClosed conn+    closed <- (||) <$> isConnectionClosed conn <*> isPeerClosed conn     sclosed <- isTxStreamClosed s     unless (receiveOnly || closed || sclosed) $ do         setTxStreamClosed s         putSendStreamQ conn $ TxStreamData s [] 0 True         waitFinTx s+    stopUnfinished s aerr     -- Outside the guard above: whether or not the FIN goes out, we are done     -- with the stream, and 'recvStream' has to say so rather than block.     setRxStreamClosed s-    delStream conn s     releaseStream s --- | Counting a stream of the peer's as done with, once, and announcing a---   new MAX_STREAMS if that is due.+-- | Asking the peer to stop sending on a stream we are done with, if it has+--   not said where the stream ends.  It then resets the stream, and what we+--   keep of it ('departStream') is settled and dropped.+stopUnfinished :: Stream -> ApplicationProtocolError -> IO ()+stopUnfinished s aerr = do+    let conn = streamConnection s+        sid = streamId s+        sendOnly =+            (isClient conn && isClientInitiatedUnidirectional sid)+                || (isServer conn && isServerInitiatedUnidirectional sid)+    closed <- (||) <$> isConnectionClosed conn <*> isPeerClosed conn+    final <- rxFinal <$> getRxBounds s+    unless (sendOnly || closed || isJust final) $ stopStream s aerr++-- | Taking the stream out of the table and counting it as done with, once. -- -- Only when the application is done with it: closing it or resetting it.  A -- RESET_STREAM from the peer takes the stream out of the table, but whatever -- is serving it may be at work yet, and counting it then would let a peer -- that opens streams and resets them straight away have any number served -- at once.+--+-- A stream of the peer's goes back to its limit on streams only once the+-- peer has finished it too.  Until then the peer may still send on it and+-- we keep what it owes; given back at once, a peer that never finishes its+-- streams could have us keep any number of them. releaseStream :: Stream -> IO () releaseStream s = do     first <- markReleased s     when first $ do-        when-            ( (isClient conn && isServerInitiatedBidirectional sid)-                || (isServer conn && isClientInitiatedBidirectional sid)-            )-            $ do-                -- FLOW CONTROL: MAX_STREAMS: recv: announcing my limit properly-                checkMaxStreams Bidirectional-        when-            ( (isClient conn && isServerInitiatedUnidirectional sid)-                || (isServer conn && isClientInitiatedUnidirectional sid)-            )-            $ do-                -- FLOW CONTROL: MAX_STREAMS: recv: announcing my limit properly-                checkMaxStreams Unidirectional-        -- FLOW CONTROL: MAX_DATA: recv: the octets of this stream the-        -- application will never read.  Left uncounted, the window we-        -- advertise stays that much smaller for the rest of the connection:-        -- a server that answers a request without reading its body pays for-        -- that body until the connection ends, and enough of them leave the-        -- peer blocked by octets nobody is waiting for.-        unread <- takeRxUnread s-        -- And, if the peer has said where the stream ends, the rest of it:-        -- RFC 9000 Sec 4.5 has a receiver account for every octet sent on a-        -- stream, and what was lost on the way was still spent.-        uncounted <- takeRxUncounted s-        let owed = unread + uncounted+        -- FLOW CONTROL: MAX_DATA: recv: what the connection counted as+        -- received for this stream and the application will never read:+        -- a request body a server answers without reading, say, or what+        -- never arrived of a stream whose end is known.  Left uncounted, the+        -- window we advertise stays that much smaller for the rest of the+        -- connection, and enough of them leave the peer blocked by octets+        -- nobody is waiting for.  Where the peer has not said where the+        -- stream ends, what arrives from here on is owed by what is kept.+        (owed, kept) <- departStream conn s         when (owed > 0) $ do             mx <- updateFlowRx conn owed-            forM_ mx $ \newMax -> do-                sendFrames conn RTT1Level [MaxData newMax]-                fire conn (Microseconds 50000) $+            case mx of+                Nothing -> return ()+                Just newMax -> do                     sendFrames conn RTT1Level [MaxData newMax]-        -- Where the peer has not said, what arrives from here on is still-        -- owed and the stream is gone, so what it owes is kept without it.-        keepDepartedStream conn s+                    fire conn (Microseconds 50000) $+                        sendFrames conn RTT1Level [MaxData newMax]+        -- FLOW CONTROL: MAX_STREAMS: recv: announcing my limit properly+        unless kept $ releasePeerStream conn $ streamId s   where     conn = streamConnection s-    sid = streamId s-    checkMaxStreams dir = do-        mx <- checkStreamIdRoom conn dir-        case mx of-            Nothing -> return ()-            Just nms -> do-                sendFrames conn RTT1Level [MaxStreams dir nms]-                fire conn (Microseconds 50000) $-                    sendFrames conn RTT1Level [MaxStreams dir nms]  -- | Accepting a stream initiated by the peer. acceptStream :: Connection -> IO Stream@@ -246,16 +250,20 @@         conn = streamConnection s     -- FLOW CONTROL: MAX_STREAM_DATA: recv: announcing my limit properly     mxs <- updateStreamFlowRx s len-    forM_ mxs $ \newMax -> do-        sendFrames conn RTT1Level [MaxStreamData sid newMax]-        fire conn (Microseconds 50000) $+    case mxs of+        Nothing -> return ()+        Just newMax -> do             sendFrames conn RTT1Level [MaxStreamData sid newMax]+            fire conn (Microseconds 50000) $+                sendFrames conn RTT1Level [MaxStreamData sid newMax]     -- FLOW CONTROL: MAX_DATA: recv: announcing my limit properly     mxc <- updateFlowRx conn len-    forM_ mxc $ \newMax -> do-        sendFrames conn RTT1Level [MaxData newMax]-        fire conn (Microseconds 50000) $+    case mxc of+        Nothing -> return ()+        Just newMax -> do             sendFrames conn RTT1Level [MaxData newMax]+            fire conn (Microseconds 50000) $+                sendFrames conn RTT1Level [MaxData newMax]     return bs  -- | Closing a stream with an error code.@@ -267,17 +275,16 @@     let sid = streamId s     sclosed <- isTxStreamClosed s     unless sclosed $ do-        finalSize <- getTxStreamFinalSize s-        setTxStreamClosed s+        finalSize <- resetTxStream s         lvl <- getEncryptionLevel conn         let frame = ResetStream sid aerr finalSize         putOutput conn $ OutControl lvl [frame]+    stopUnfinished s aerr     -- Outside the guard above: whether or not the RESET_STREAM goes out, we     -- are done with the stream, and 'recvStream' has to say so rather than     -- block.  The peer's STOP_SENDING has closed the sending part already     -- when an application resets a stream because 'sendStream' failed.     setRxStreamClosed s-    delStream conn s     releaseStream s  -- | Asking the peer to stop sending.@@ -334,4 +341,4 @@  -- | Receiving a DATAGRAM frame in STM. recvDatagramSTM :: Connection -> STM ByteString-recvDatagramSTM = readTQueue . connRecvDatagramQ+recvDatagramSTM conn = readTQueue (connRecvDatagramQ conn) `orElse` peerClosedSTM conn

@@ -14,6 +14,7 @@     module Network.QUIC.Utils,     module Network.QUIC.Recovery,     module Network.QUIC.Client.Reader,+    module Network.QUIC.Client.WatchDog,     module Network.QUIC.Socket, ) where @@ -23,6 +24,7 @@     readerClient,     rebind,  )+import Network.QUIC.Client.WatchDog (watchDogWith, watchSocket) import Network.QUIC.Config import Network.QUIC.Connection import Network.QUIC.Connector

@@ -78,7 +78,10 @@                 bytePN = bsXOR mask2 epn                 headerLen = cryptPktNumOffset + epnLen                 (proHeader, ciphertext) = BS.splitAt headerLen cryptPacket-            peerPN <- if lvl == RTT1Level then getPeerPacketNumber conn else return 0+            -- RFC 9000 Sec 17.1 and A.3: against the largest in the same+            -- space.  Against 0, a truncated number past 255 came out wrong+            -- and the packet failed to decrypt.+            peerPN <- getPeerPacketNumber conn lvl             let pn = decodePacketNumber peerPN (toEncodedPacketNumber bytePN) epnLen             header <- BS.create headerLen $ \p -> do                 void $ copy p proHeader

@@ -322,8 +322,8 @@  ---------------------------------------------------------------- --- | A Stateless Reset carrying the given token: 1280 octets, of which the---   last sixteen are the token and the rest is noise.+-- | A Stateless Reset of the given size carrying the given token: the+--   last sixteen octets are the token and the rest is noise. -- -- RFC 9000 Sec 10.3 fixes the first two bits at 01 -- header form 0 and the -- QUIC Bit set -- so that it cannot be told from an ordinary short header@@ -333,15 +333,15 @@ -- in it.  We are answering a packet we have no connection for, so whether the -- peer asked is exactly what we cannot know. ----- 1280 octets is under three times the 428 the caller insists on having--- received, which is what Sec 10.3 allows to be sent in answer.-makeStatelessReset :: StatelessResetToken -> IO ByteString-makeStatelessReset srt = do+-- The caller chooses the size with 'statelessResetSizeFor'.+makeStatelessReset :: Int -> StatelessResetToken -> IO ByteString+makeStatelessReset siz srt = do     r <- randomRIO (0, 255)     let flag = 0x40 .|. (r .&. 0x3f)+        bodySize = siz - 1 - 16 #if MIN_VERSION_random(1,3,0)-    body <- getStdRandom $ uniformByteString 1263+    body <- getStdRandom $ uniformByteString bodySize #else-    body <- getStdRandom $ genByteString 1263+    body <- getStdRandom $ genByteString bodySize #endif     return $ BS.concat [BS.singleton flag, body, fromStatelessResetToken srt]

@@ -6,6 +6,7 @@     countZero, -- testing ) where +import qualified Control.Exception as E import qualified Data.ByteString as BS import qualified Data.ByteString.Short as Short import Foreign.Ptr (Ptr, alignPtr, castPtr, minusPtr, plusPtr)@@ -41,10 +42,10 @@ encodeFrame :: WriteBuffer -> Frame -> IO () encodeFrame wbuf (Padding n) = replicateM_ n $ write8 wbuf 0x00 encodeFrame wbuf Ping = write8 wbuf 0x01-encodeFrame wbuf (Ack (AckInfo largest range1 ranges) (Milliseconds delay)) = do+encodeFrame wbuf (Ack (AckInfo largest range1 ranges) delay) = do     write8 wbuf 0x02     encodeInt' wbuf $ fromIntegral largest-    encodeInt' wbuf $ fromIntegral delay+    encodeDelay wbuf delay     encodeInt' wbuf $ fromIntegral $ length ranges     encodeInt' wbuf $ fromIntegral range1     mapM_ putRanges ranges@@ -159,8 +160,12 @@ decodeFramesBuffer :: Buffer -> BufferSize -> IO (Maybe [Frame]) decodeFramesBuffer buf bufsiz = newReadBuffer buf bufsiz >>= decodeFrames +-- | 'Nothing' for a frame of unknown type, or one cut short.  The latter+--   ran the decoder past the end of the packet, and the exception it threw+--   closed the connection with INTERNAL_ERROR rather than+--   FRAME_ENCODING_ERROR (RFC 9000 Sec 12.4). decodeFrames :: ReadBuffer -> IO (Maybe [Frame])-decodeFrames rbuf = loop id+decodeFrames rbuf = E.handle (\BufferOverrun -> return Nothing) $ loop id   where     loop frames = do         ok <- (>= 1) <$> remainingSize rbuf@@ -262,11 +267,11 @@ decodeAck :: ReadBuffer -> IO Frame decodeAck rbuf = do     largest <- fromIntegral <$> decodeInt' rbuf-    delay <- fromIntegral <$> decodeInt' rbuf+    delay <- decodeDelay rbuf     count <- fromIntegral <$> decodeInt' rbuf     range1 <- fromIntegral <$> decodeInt' rbuf     ranges <- getRanges count id-    return $ Ack (AckInfo largest range1 ranges) $ Milliseconds delay+    return $ Ack (AckInfo largest range1 ranges) delay   where     getRanges 0 build = return $ build []     getRanges n build = do

@@ -7,7 +7,7 @@  import qualified Control.Exception as E import qualified Data.ByteString as BS-import Network.Control+import Network.Socket (SockAddr) import Network.TLS (AlertDescription (..)) import System.Log.FastLogger @@ -57,6 +57,8 @@         rpkt <- recvTimeout         let CryptPacket hdr _ = rpCryptPacket rpkt             cid = headerMyCID hdr+        -- Before 'setMyCID' below, which makes it the one in use.+        cidChanged <- (/= cid) <$> getMyCID conn         included <- myCIDsInclude conn cid         case included of             Just nseq -> do@@ -77,7 +79,7 @@                     writeIORef (sentRetirePriorTo conn) False                     unless sent $                         sendFrames conn RTT1Level [NewConnectionID cidInfo 0]-                processReceivedPacket conn rpkt+                processReceivedPacket' conn (Just cidChanged) rpkt                 shouldUpdatePeer <-                     if shouldUpdate                         then shouldUpdatePeerCID conn@@ -140,19 +142,13 @@                         onPacketNumberSpaceDiscarded ldcc InitialLevel                 processReceivedPacket conn rpkt -rateLimit :: Int-rateLimit = 32--checkRate :: [Frame] -> Int-checkRate fs0 = go fs0 0-  where-    go [] n = n-    go (f : fs) n-        | rateControled f = go fs (n + 1)-        | otherwise = go fs n- processReceivedPacket :: Connection -> ReceivedPacket -> IO ()-processReceivedPacket conn rpkt = do+processReceivedPacket conn = processReceivedPacket' conn Nothing++-- | With whether the packet came on a CID other than the one in use, once+--   the handshake is over: only then may the peer migrate.+processReceivedPacket' :: Connection -> Maybe Bool -> ReceivedPacket -> IO ()+processReceivedPacket' conn mcidChanged rpkt = do     let CryptPacket hdr crypt = rpCryptPacket rpkt         lvl = rpEncryptionLevel rpkt         tim = rpTimeRecevied rpkt@@ -175,15 +171,12 @@             when (isIllegalReservedBits plainMarks || isNoFrames plainMarks) $                 closeConnection conn ProtocolViolation "Non 0 RR bits or no frames"             when (isUnknownFrame plainMarks) $-                closeConnection conn FrameEncodingError "Unknown frame"-            let controlled = checkRate plainFrames-            when (controlled /= 0) $ do-                rate <- addRate (controlRate conn) controlled-                when (rate > rateLimit) $ do-                    closeConnection conn QUIC.InternalError "Rate control"+                closeConnection conn FrameEncodingError "Unknown or truncated frame"+            enforcePathAndCIDFrameLimit conn plainFrames             -- For Ping, record PPN first, then send an ACK.             onPacketReceived (connLDCC conn) lvl plainPacketNumber-            when (lvl == RTT1Level) $ setPeerPacketNumber conn plainPacketNumber+            largest <- (plainPacketNumber >) <$> getPeerPacketNumber conn lvl+            setPeerPacketNumber conn lvl plainPacketNumber             qlogReceived conn (PlainPacket hdr plain) tim             let ackEli = any ackEliciting plainFrames             (ckp, cpn) <- getCurrentKeyPhase conn@@ -194,6 +187,14 @@                 updateCoder1RTT conn ckp -- ckp is now next             checkRetireOfArrivalCID conn hdr plainFrames             mapM_ (processFrame conn lvl) plainFrames+            -- What the frames above made us queue, a NEW_CONNECTION_ID+            -- flood in particular, must not pile up for good.+            enforceQueuedControlLimit conn+            case (mcidChanged, rpPeerAddr rpkt) of+                (Just cidChanged, Just peersa)+                    | largest && any (not . isProbing) plainFrames ->+                        checkMigration conn (headerMyCID hdr) peersa cidChanged+                _ -> return ()             when ackEli $ do                 case lvl of                     RTT0Level -> return ()@@ -240,8 +241,9 @@  -- | Closing the connection over a final size that does not hold together --   (RFC 9000 Sec 4.5).-closeOverFinalSize :: Connection -> Maybe FinalSizeProblem -> IO ()-closeOverFinalSize conn = mapM_ $ closeConnection conn FinalSizeError . reason+closeOverFinalSize :: Connection -> FinalSizeProblem -> IO ()+closeOverFinalSize conn problem =+    closeConnection conn FinalSizeError $ reason problem   where     reason FinalSizeChanged = "a final size that is not the one already known"     reason FinalSizeTooSmall = "a final size below what the stream has reached"@@ -307,6 +309,56 @@ -- addressed to another.  A retransmission cannot either -- it goes to a -- connection ID we still have, and the retired one we no longer answer on -- at all.+-- | RFC 9000 Sec 9.3: the peer has moved when a packet that decrypts,+--   carries a non-probing frame and is the largest so far comes from a new+--   address.  Nothing before decryption: anyone who has seen the CID can+--   send a packet from anywhere.+checkMigration :: Connection -> CID -> SockAddr -> Bool -> IO ()+checkMigration conn dCID peersa cidChanged = do+    mPathInfo <- findPathInfo conn peersa+    case mPathInfo of+        Just _ -> return ()+        Nothing -> do+            -- Claimed before the thread rather than inside it, so that a+            -- peer moving gets one thread for the migration and not one+            -- for every packet.+            claimed <- claimMigration conn+            when claimed $+                forkManaged conn "QUIC migration" $+                    onClientMigration conn dCID peersa cidChanged++-- | Frames that do not move the peer (RFC 9000 Sec 9.1).+isProbing :: Frame -> Bool+isProbing Padding{} = True+isProbing PathChallenge{} = True+isProbing PathResponse{} = True+isProbing NewConnectionID{} = True+isProbing _ = False++-- | Carry out the migration the caller has claimed with 'claimMigration'.+onClientMigration :: Connection -> CID -> SockAddr -> Bool -> IO ()+onClientMigration conn newdCID peersa cidChanged = handleLogUnit logAction $ do+    -- fixme: should not block+    mcidinfo <-+        if cidChanged+            then timeout (Microseconds 100000) "onClientMigration" $ waitPeerCID conn+            else return Nothing -- PathChallenge only, no RetireConnectionID+    let msg = "Migration: " <> bhow peersa <> " (" <> bhow newdCID <> ")"+    qlogDebug conn $ Debug $ toLogStr msg+    connDebugLog conn $ "debug: onClientMigration: " <> msg+    oldPathInfo <- getPathInfo conn+    pathInfo <- newPathInfo peersa+    -- assumed that this PathInfo is not stored in PeerInfo+    addPathInfo conn pathInfo+    ok <- validatePath conn pathInfo mcidinfo+    -- RFC 9000 Sec 9.3.2: back to the last validated address.  The one+    -- that failed is forgotten, so that a packet from it starts afresh.+    unless ok $ do+        connDebugLog conn $ "debug: onClientMigration: no response from " <> bhow peersa+        restorePathInfo conn oldPathInfo+  where+    logAction msg = connDebugLog conn ("debug: onClientMigration: " <> msg)+ checkRetireOfArrivalCID :: Connection -> Header -> [Frame] -> IO () checkRetireOfArrivalCID conn hdr frames = do     mseq <- myCIDsInclude conn $ headerMyCID hdr@@ -321,17 +373,103 @@   where     retired = [n | RetireConnectionID n <- frames] --- | Giving the connection's window back what a frame for a stream the---   application has closed is owed.-creditDeparted :: Connection -> StreamId -> Int -> Bool -> IO ()-creditDeparted conn sid end fin = do-    owed <- noteDepartedRxFrame conn sid end fin-    when (owed > 0) $ do-        mx <- updateFlowRx conn owed-        forM_ mx $ \newMax -> do+-- | Counting as received what a frame took a stream past what had been+--   seen of it, and refusing it if that is past the connection's limit.+--+-- RFC 9000 Sec 4.1 has the octets received be "the sum of the largest+-- received offsets on all streams", which is what the peer counts as spent.+-- They were the octets reassembled instead, so a stream's gaps -- what the+-- peer sent and we never saw -- were counted by nobody, and a peer could go+-- past its limit by them unnoticed.+countedRx :: Connection -> Either FinalSizeProblem Int -> IO Int+countedRx conn r = case r of+    Left problem -> do+        closeOverFinalSize conn problem+        return 0 -- not reached: closeConnection throws+    Right n -> do+        -- FLOW CONTROL: MAX_DATA: recv: rejecting if over my limit+        when (n > 0) $ do+            ok <- checkRxMaxData conn n+            unless ok $+                closeConnection conn FlowControlError "Flow control error for connection"+        return n++-- | Counting octets nobody will read as consumed, and announcing a new+--   MAX_DATA if that is due.+giveBack :: Connection -> Int -> IO ()+giveBack conn n = when (n > 0) $ do+    mx <- updateFlowRx conn n+    case mx of+        Nothing -> return ()+        Just newMax -> do             sendFrames conn RTT1Level [MaxData newMax]             fire conn (Microseconds 50000) $ sendFrames conn RTT1Level [MaxData newMax] +-- | A frame for a stream the application has closed: counted as received+--   and, since nobody will read it, as consumed, and the stream given back+--   to the peer's limit once it has ended.+creditDeparted+    :: Connection+    -> StreamId+    -> Either (TransportError, ReasonPhrase) (Int, Bool)+    -> IO ()+creditDeparted conn sid r =+    case r of+        Left (err, why) -> closeConnection conn err why+        Right (n, done) -> do+            -- FLOW CONTROL: MAX_DATA: recv: rejecting if over my limit+            when (n > 0) $ do+                ok <- checkRxMaxData conn n+                unless ok $+                    closeConnection conn FlowControlError "Flow control error for connection"+            giveBack conn n+            when done $ releasePeerStream conn sid++-- | A STREAM frame whose stream id has been looked over.+receiveStreamFrame+    :: Connection -> ReasonPhrase -> StreamId -> Offset -> ByteString -> Bool -> IO ()+receiveStreamFrame conn space sid off dat fin = do+    -- FLOW CONTROL: MAX_DATA: recv: counted by how far it takes the stream,+    -- before the reassembly drops what it has seen already.+    a <- noteRxArrival conn sid end fin False+    case a of+        ArrivedLive strm r -> take' (Just strm) strm r+        -- The stream is closed and the data goes nowhere, but the peer spent+        -- its connection window on it and is owed that back.+        ArrivedDeparted r -> creditDeparted conn sid r+        ArrivedNowhere -> do+            guardStream conn sid Nothing+            -- Nothing when the stream has been closed.  What arrives for it+            -- now is a copy of something already received, sent again+            -- because our acknowledgement crossed it.  Opening the stream+            -- anew would hold the copy to the initial window and call it a+            -- flow control error.+            mstrm <- openStream conn sid+            case mstrm of+                Nothing -> return ()+                Just strm -> noteRxFrame strm end fin >>= take' Nothing strm+  where+    len = BS.length dat+    end = off + len+    take' found strm r = do+        -- RFC 9000 Sec 4.5: where the stream ends, once said, cannot be+        -- said differently, and nothing may arrive past it.+        _ <- countedRx conn r+        fc <- putRxStreamData strm $ RxStreamData dat off len fin+        case fc of+            -- FLOW CONTROL: MAX_STREAM_DATA: recv: rejecting if over my limit+            OverLimit ->+                closeConnection conn FlowControlError $+                    "Flow control error for stream in " <> space+            -- Not a flow control error: the peer is inside its window, it is+            -- just spending it in more pieces than we will hold.  Rate+            -- control answers with InternalError too.+            TooFragmented ->+                closeConnection conn QUIC.InternalError "Too many stream fragments"+            Duplicated -> return ()+            Reassembled _ -> return ()+        deliverStream conn found strm+ processFrame :: Connection -> EncryptionLevel -> Frame -> IO () processFrame _ _ Padding{} = return () processFrame conn lvl Ping = do@@ -353,7 +491,8 @@     nextPN <- getPacketNumber conn     when (largestAcked >= nextPN) $         closeConnection conn ProtocolViolation "ACK for a packet never sent"-    onAckReceived (connLDCC conn) lvl ackInfo $ milliToMicro ackDelay+    e <- ackDelayExponent <$> getPeerParameters conn+    onAckReceived (connLDCC conn) lvl ackInfo $ ackDelayMicro e ackDelay processFrame conn lvl (ResetStream sid aerr finlen) = do     when (lvl == InitialLevel || lvl == HandshakeLevel) $         closeConnection conn ProtocolViolation "RESET_STREAM"@@ -363,8 +502,6 @@     -- FLOW CONTROL: MAX_STREAMS: recv: rejecting if over my limit     ok <- checkRxMaxStreams conn sid     unless ok $ closeConnection conn StreamLimitError "stream id is too large"-    mstrm0 <- findStream conn sid-    guardStream conn sid mstrm0     -- RFC 9000 Sec 3.2: "The receiving part of a stream initiated by a peer     -- ... is created when the first STREAM, STREAM_DATA_BLOCKED, or     -- RESET_STREAM frame is received for that stream."  Only a STREAM frame@@ -377,64 +514,75 @@     -- then said its piece and been ignored, and ours waited on 'recvStream'     -- for a stream that would never end, until the idle timeout.     ---    -- 'openStream' answers Nothing for a stream that has been closed since,-    -- so a late copy does not open one anew.-    mstrm <- maybe (openStream conn sid) (return . Just) mstrm0-    onResetStreamReceived2 (connHooks conn) mstrm aerr finlen-    case mstrm of+    -- FLOW CONTROL: MAX_DATA: recv: what the final size takes the stream+    -- past will never arrive, and nobody will read it, so it is given back+    -- at once.  Left to the close, a peer resetting streams the application+    -- is slow to close is kept waiting.+    a <- noteRxArrival conn sid finlen True True+    case a of+        ArrivedLive strm r -> reset (Just strm) strm r         -- As for a STREAM frame arriving for a stream the application has         -- closed: this says where the stream ends, so what the peer spent on         -- it is known in full and owed in full.-        Nothing -> creditDeparted conn sid finlen True-        Just strm -> do-            -- RFC 9000 Sec 4.5, as for a STREAM frame that ends the stream.-            noteRxFinalSize strm finlen >>= closeOverFinalSize conn-            -- FLOW CONTROL: MAX_DATA: recv: the octets the peer spent on the-            -- stream that will now never arrive.  Left uncounted, the window-            -- we advertise falls behind what the peer believes it has spent,-            -- by the tail of every stream it resets, until it has none left.-            unarrived <- takeRxUncounted strm-            when (unarrived > 0) $ do-                ok' <- checkRxMaxData conn unarrived-                unless ok' $-                    closeConnection conn FlowControlError "Flow control error for connection"-                mx <- updateFlowRx conn unarrived-                forM_ mx $ \newMax -> do-                    sendFrames conn RTT1Level [MaxData newMax]-                    fire conn (Microseconds 50000) $-                        sendFrames conn RTT1Level [MaxData newMax]-            -- After the two checks above, not before: the application has no-            -- business serving a stream the frame that opened it is about to-            -- close the connection over.  As for a STREAM frame (#131).-            deliverStream conn mstrm0 strm-            onResetStreamReceived (connHooks conn) strm aerr-            -- Before the pseudo FIN below, so that whoever reads it can-            -- tell it was not a real one.-            setResetReceived strm aerr-            -- RESET_STREAM ends the peer's sending part alone-            -- (RFC 9000, section 3.2).  Our own sending part is untouched,-            -- so that a reply to what the peer sent before the reset can-            -- still go out.  It used to be closed here as well, and-            -- 'sendStream' then threw StreamIsClosed.-            setRxStreamClosed strm-            delStream conn strm+        ArrivedDeparted r -> do+            onResetStreamReceived2 (connHooks conn) Nothing aerr finlen+            creditDeparted conn sid r+        ArrivedNowhere -> do+            guardStream conn sid Nothing+            -- 'openStream' answers Nothing for a stream that has been closed+            -- since, so a late copy does not open one anew.+            mstrm <- openStream conn sid+            case mstrm of+                Nothing -> onResetStreamReceived2 (connHooks conn) Nothing aerr finlen+                Just strm -> do+                    r <- noteRxFinalSize strm finlen+                    case r of+                        Right n -> addRxCredited strm n+                        Left _ -> return ()+                    reset Nothing strm r+  where+    reset found strm r = do+        onResetStreamReceived2 (connHooks conn) (Just strm) aerr finlen+        -- FLOW CONTROL: MAX_STREAM_DATA: recv: rejecting if over my limit.+        -- The final size is what the peer spent on the stream, and it was+        -- taken however far past the stream's limit it was.+        lim <- getRxMaxStreamData strm+        when (finlen > lim) $+            closeConnection conn FlowControlError "Flow control error for stream"+        -- RFC 9000 Sec 4.5, as for a STREAM frame that ends the stream.+        countedRx conn r >>= giveBack conn+        -- After the two checks above, not before: the application has no+        -- business serving a stream the frame that opened it is about to+        -- close the connection over.  As for a STREAM frame (#131).+        deliverStream conn found strm+        onResetStreamReceived (connHooks conn) strm aerr+        -- Before the pseudo FIN below, so that whoever reads it can+        -- tell it was not a real one.+        setResetReceived strm aerr+        -- RESET_STREAM ends the peer's sending part alone+        -- (RFC 9000, section 3.2).  Our own sending part is untouched,+        -- so that a reply to what the peer sent before the reset can+        -- still go out.  It used to be closed here as well, and+        -- 'sendStream' then threw StreamIsClosed.+        setRxStreamClosed strm+        delStream conn strm processFrame conn lvl (StopSending sid err) = do-    ok <- rateOK conn-    when ok $ do-        when (lvl == InitialLevel || lvl == HandshakeLevel) $-            closeConnection conn ProtocolViolation "STOP_SENDING"-        when (isReceiveOnly conn sid) $-            closeConnection conn StreamStateError "Receive-only stream"-        mstrm <- findStream conn sid-        case mstrm of-            Nothing -> streamNotCreatedYet conn sid "No such stream for STOP_SENDING"-            Just strm -> do-                finalSize <- getTxStreamFinalSize strm-                sendFrames conn lvl [ResetStream sid err finalSize]-                -- Our sending part is reset now (RFC 9000, section 3.5), so-                -- no more STREAM frames may go out on it.  Without this,+    when (lvl == InitialLevel || lvl == HandshakeLevel) $+        closeConnection conn ProtocolViolation "STOP_SENDING"+    when (isReceiveOnly conn sid) $+        closeConnection conn StreamStateError "Receive-only stream"+    mstrm <- findStream conn sid+    case mstrm of+        Nothing -> streamNotCreatedYet conn sid "No such stream for STOP_SENDING"+        Just strm -> do+            -- Once per stream, however often the peer asks: a reset sending+            -- part has said all it will (RFC 9000, section 3.5).+            reset <- isTxStreamReset strm+            unless reset $ do+                -- No more STREAM frames may go out on it.  Without this,                 -- 'sendStream' went on working after the RESET_STREAM.-                setTxStreamClosed strm+                finalSize <- resetTxStream strm+                sendFrames conn lvl [ResetStream sid err finalSize] processFrame _ _ (CryptoF _ "") = return () processFrame conn lvl (CryptoF off cdat) = do     when (lvl == RTT0Level) $@@ -466,40 +614,7 @@     unless ok $ closeConnection conn StreamLimitError "stream id is too large"     when (isSendOnly conn sid) $         closeConnection conn StreamStateError "send-only stream"-    mstrm <- findStream conn sid-    guardStream conn sid mstrm-    -- Nothing when the stream has been closed.  What arrives for it now-    -- is a copy of something already received, sent again because our-    -- acknowledgement crossed it.  Opening the stream anew would hold-    -- the copy to the initial window and call it a flow control error.-    mstrm' <- maybe (openStream conn sid) (return . Just) mstrm-    forM_ mstrm' $ \strm -> do-        let len = BS.length dat-            rx = RxStreamData dat off len fin-        -- RFC 9000 Sec 4.5: where the stream ends, once said, cannot be said-        -- differently, and nothing may arrive past it.-        noteRxFrame strm (off + len) fin >>= closeOverFinalSize conn-        fc <- putRxStreamData strm rx-        case fc of-            -- FLOW CONTROL: MAX_STREAM_DATA: recv: rejecting if over my limit-            OverLimit ->-                closeConnection conn FlowControlError "Flow control error for stream in 0-RTT"-            -- Not a flow control error: the peer is inside its window, it is-            -- just spending it in more pieces than we will hold.  Rate control-            -- answers with InternalError too.-            TooFragmented ->-                closeConnection conn QUIC.InternalError "Too many stream fragments"-            Duplicated -> return ()-            Reassembled -> do-                addRxCounted strm len-                ok' <- checkRxMaxData conn len-                -- FLOW CONTROL: MAX_DATA: send: respecting peer's limit-                unless ok' $-                    closeConnection-                        conn-                        FlowControlError-                        "Flow control error for connection in 0-RTT"-        deliverStream conn mstrm strm+    receiveStreamFrame conn "0-RTT" sid off dat fin processFrame conn RTT1Level (StreamF sid _ [""] False) = do     -- FLOW CONTROL: MAX_STREAMS: recv: rejecting if over my limit     ok <- checkRxMaxStreams conn sid@@ -515,44 +630,7 @@     unless ok $ closeConnection conn StreamLimitError "stream id is too large"     when (isSendOnly conn sid) $         closeConnection conn StreamStateError "send-only stream"-    mstrm <- findStream conn sid-    guardStream conn sid mstrm-    -- Nothing when the stream has been closed.  What arrives for it now-    -- is a copy of something already received, sent again because our-    -- acknowledgement crossed it.  Opening the stream anew would hold-    -- the copy to the initial window and call it a flow control error.-    mstrm' <- maybe (openStream conn sid) (return . Just) mstrm-    case mstrm' of-        -- The stream is closed and the data goes nowhere, but the peer spent-        -- its connection window on it and is owed that back.-        Nothing -> creditDeparted conn sid (off + BS.length dat) fin-        Just strm -> do-            let len = BS.length dat-                rx = RxStreamData dat off len fin-            -- RFC 9000 Sec 4.5: where the stream ends, once said, cannot be-            -- said differently, and nothing may arrive past it.-            noteRxFrame strm (off + len) fin >>= closeOverFinalSize conn-            fc <- putRxStreamData strm rx-            case fc of-                -- FLOW CONTROL: MAX_STREAM_DATA: recv: rejecting if over my limit-                OverLimit ->-                    closeConnection conn FlowControlError "Flow control error for stream in 1-RTT"-                -- Not a flow control error: the peer is inside its window, it-                -- is just spending it in more pieces than we will hold.  Rate-                -- control answers with InternalError too.-                TooFragmented ->-                    closeConnection conn QUIC.InternalError "Too many stream fragments"-                Duplicated -> return ()-                Reassembled -> do-                    addRxCounted strm len-                    ok' <- checkRxMaxData conn len-                    -- FLOW CONTROL: MAX_DATA: send: respecting peer's limit-                    unless ok' $-                        closeConnection-                            conn-                            FlowControlError-                            "Flow control error for connection in 1-RTT"-            deliverStream conn mstrm strm+    receiveStreamFrame conn "1-RTT" sid off dat fin processFrame conn lvl (MaxData n) = do     when (lvl == InitialLevel || lvl == HandshakeLevel) $         closeConnection conn ProtocolViolation "MAX_DATA in Initial or Handshake"@@ -598,7 +676,9 @@     -- thing to hear only when we have told it nothing may be sent -- an     -- initial_max_stream_data of zero for that kind of stream.     mstrm <- maybe (openStream conn sid) (return . Just) mstrm0-    forM_ mstrm $ deliverStream conn mstrm0+    case mstrm of+        Nothing -> return ()+        Just strm -> deliverStream conn mstrm0 strm processFrame conn lvl (StreamsBlocked _dir n) = do     when (lvl == InitialLevel || lvl == HandshakeLevel) $         closeConnection conn ProtocolViolation "STREAMS_BLOCKED in Initial or Handshake"@@ -610,62 +690,60 @@             conn             ProtocolViolation             "NEW_CONNECTION_ID in Initial or Handshake"-    ok <- rateOK conn-    when ok $ do-        let (_, cidlen) = unpackCID $ cidInfoCID cidInfo-            seqNum = cidInfoSeq cidInfo-        when (cidlen < 1 || 20 < cidlen || retirePriorTo > seqNum) $-            closeConnection conn FrameEncodingError "NEW_CONNECTION_ID parameter error"-        -- RFC 9000 Sec 19.15 says:-        -- If an endpoint receives a NEW_CONNECTION_ID frame that repeats a-        -- previously issued connection ID with a different Stateless Reset-        -- Token field value or a different Sequence Number field value, or-        -- if a sequence number is used for different connection IDs, the-        -- endpoint MAY treat that receipt as a connection error of type-        -- PROTOCOL_VIOLATION.-        ---        -- Before the retirement below: a frame that contradicts what the-        -- peer has already said is not one to act on.-        consistent <- isPeerCIDConsistent conn cidInfo-        unless consistent $-            closeConnection conn ProtocolViolation "NEW_CONNECTION_ID repeated differently"-        -- Retiring CIDs first then add a new CID.-        ---        -- RFC 9000 Sec 5.1.1 says:-        -- An endpoint MAY send connection IDs that temporarily exceed a-        -- peer's limit if the NEW_CONNECTION_ID frame also requires the-        -- retirement of any excess, by including a sufficiently large-        -- value in the Retire Prior To field.-        prevRetirePriorTo <- getPeerRetirePriorTo conn-        -- RFC 900 Sec 19.15 says:-        -- Once a sender indicates a Retire Prior To value, smaller values-        -- sent in subsequent NEW_CONNECTION_ID frames have no effect. A-        -- receiver MUST ignore any Retire Prior To fields that do not-        -- increase the largest received Retire Prior To value.-        when (retirePriorTo >= prevRetirePriorTo) $ do-            -- RFC 9000 Sec 5.1.2 says:-            -- Upon receipt of an increased Retire Prior To field, the-            -- peer MUST stop using the corresponding connection IDs and-            -- retire them with RETIRE_CONNECTION_ID frames before adding-            -- the newly provided connection ID to the set of active-            -- connection IDs.-            seqNums <- setPeerCIDAndRetireCIDs conn retirePriorTo -- upadting RPT-            sendFrames conn RTT1Level $ map RetireConnectionID seqNums-        -- Adding a new CID-        if seqNum < prevRetirePriorTo-            then-                -- RFC 9000 Sec 19.15 says:-                -- An endpoint that receives a NEW_CONNECTION_ID frame-                -- with a sequence number smaller than the Retire Prior To-                -- field of a previously received NEW_CONNECTION_ID frame-                -- MUST send a corresponding RETIRE_CONNECTION_ID frame-                -- that retires the newly received connection ID, unless-                -- it has already done so for that sequence number.-                sendFrames conn RTT1Level [RetireConnectionID seqNum]-            else do-                ng <- not <$> addPeerCID conn cidInfo-                when ng $-                    closeConnection conn ConnectionIdLimitError "NEW_CONNECTION_ID limit error"+    let (_, cidlen) = unpackCID $ cidInfoCID cidInfo+        seqNum = cidInfoSeq cidInfo+    when (cidlen < 1 || 20 < cidlen || retirePriorTo > seqNum) $+        closeConnection conn FrameEncodingError "NEW_CONNECTION_ID parameter error"+    -- RFC 9000 Sec 19.15 says:+    -- If an endpoint receives a NEW_CONNECTION_ID frame that repeats a+    -- previously issued connection ID with a different Stateless Reset+    -- Token field value or a different Sequence Number field value, or+    -- if a sequence number is used for different connection IDs, the+    -- endpoint MAY treat that receipt as a connection error of type+    -- PROTOCOL_VIOLATION.+    --+    -- Before the retirement below: a frame that contradicts what the+    -- peer has already said is not one to act on.+    consistent <- isPeerCIDConsistent conn cidInfo+    unless consistent $+        closeConnection conn ProtocolViolation "NEW_CONNECTION_ID repeated differently"+    -- Retiring CIDs first then add a new CID.+    --+    -- RFC 9000 Sec 5.1.1 says:+    -- An endpoint MAY send connection IDs that temporarily exceed a+    -- peer's limit if the NEW_CONNECTION_ID frame also requires the+    -- retirement of any excess, by including a sufficiently large+    -- value in the Retire Prior To field.+    prevRetirePriorTo <- getPeerRetirePriorTo conn+    -- RFC 900 Sec 19.15 says:+    -- Once a sender indicates a Retire Prior To value, smaller values+    -- sent in subsequent NEW_CONNECTION_ID frames have no effect. A+    -- receiver MUST ignore any Retire Prior To fields that do not+    -- increase the largest received Retire Prior To value.+    when (retirePriorTo >= prevRetirePriorTo) $ do+        -- RFC 9000 Sec 5.1.2 says:+        -- Upon receipt of an increased Retire Prior To field, the+        -- peer MUST stop using the corresponding connection IDs and+        -- retire them with RETIRE_CONNECTION_ID frames before adding+        -- the newly provided connection ID to the set of active+        -- connection IDs.+        seqNums <- setPeerCIDAndRetireCIDs conn retirePriorTo -- upadting RPT+        sendFrames conn RTT1Level $ map RetireConnectionID seqNums+    -- Adding a new CID+    if seqNum < prevRetirePriorTo+        then+            -- RFC 9000 Sec 19.15 says:+            -- An endpoint that receives a NEW_CONNECTION_ID frame+            -- with a sequence number smaller than the Retire Prior To+            -- field of a previously received NEW_CONNECTION_ID frame+            -- MUST send a corresponding RETIRE_CONNECTION_ID frame+            -- that retires the newly received connection ID, unless+            -- it has already done so for that sequence number.+            sendFrames conn RTT1Level [RetireConnectionID seqNum]+        else do+            ng <- not <$> addPeerCID conn cidInfo+            when ng $+                closeConnection conn ConnectionIdLimitError "NEW_CONNECTION_ID limit error" processFrame conn RTT1Level (RetireConnectionID sn) = do     -- RFC 9000 Sec 19.16: "Receipt of a RETIRE_CONNECTION_ID frame     -- containing a sequence number greater than any previously sent to the@@ -690,13 +768,14 @@                 unregister <- getUnregister conn                 unregister $ cidInfoCID cidInfo processFrame conn RTT1Level (PathChallenge dat) = do-    ok <- rateOK conn+    ok <- withinPathResponseLimit conn     when ok $ sendFrames conn RTT1Level [PathResponse dat] processFrame conn RTT1Level (PathResponse dat) =     -- RTT0Level falls intentionally     checkResponse conn dat-processFrame conn lvl (ConnectionClose NoError _ftyp reason) =-    when (isServer conn || lvl /= RTT1Level) $ E.throwIO $ ConnectionIsClosed reason+processFrame conn _lvl (ConnectionClose NoError _ftyp reason) = do+    setPeerClosed conn reason+    E.throwIO $ ConnectionIsClosed reason processFrame _conn _lvl (ConnectionClose err _ftyp reason) = do     let quicexc = TransportErrorIsReceived err reason     E.throwIO quicexc@@ -752,7 +831,7 @@                     closeConnection conn QUIC.InternalError "Too many CRYPTO fragments"                     return False -- not reached: closeConnection throws                 Duplicated -> return True-                Reassembled -> return False+                Reassembled _ -> return False  killHandshaker :: Connection -> EncryptionLevel -> IO () killHandshaker conn lvl = putCrypto conn $ InpHandshake lvl ""

@@ -98,11 +98,13 @@ cancelLossDetectionTimer ldcc@LDCC{..} = do     atomically $ writeTVar timerInfoQ Empty     mk <- atomicModifyIORef' timerKey (Nothing,)-    forM_ mk $ \k -> do-        mgr <- getSystemTimerManager-        unregisterTimeout mgr k-        writeIORef timerInfo Nothing-        qlogLossTimerCancelled ldcc+    case mk of+        Nothing -> return ()+        Just k -> do+            mgr <- getSystemTimerManager+            unregisterTimeout mgr k+            writeIORef timerInfo Nothing+            qlogLossTimerCancelled ldcc  updateLossDetectionTimer :: LDCC -> TimerInfo -> IO () updateLossDetectionTimer ldcc@LDCC{..} tmi = do@@ -142,7 +144,9 @@         mgr <- getSystemTimerManager         key <- registerTimeout mgr us (onLossDetectionTimeout ldcc)         mk <- atomicModifyIORef' timerKey (Just key,)-        forM_ mk $ unregisterTimeout mgr+        case mk of+            Nothing -> return ()+            Just k -> unregisterTimeout mgr k         writeIORef timerInfo $ Just tmi  ----------------------------------------------------------------

@@ -203,7 +203,9 @@                     Nothing -> do                         qlogDebug conn $ Debug "probe ping"                         return [Ping]-        xs <- construct conn lvl frames False+        -- A probe must elicit an ACK, whatever was left of the packet.+        let frames' = if any ackEliciting frames then frames else Ping : frames+        xs <- construct conn lvl frames' False         -- Asking for the last one and asking whether there is one at all are         -- the same question, so ask it once.         case reverse xs of@@ -295,8 +297,15 @@         ppkt = PlainPacket header plain     return [mkSentPacket mypn lvl ppkt ppns ackEli]   where-    mkAck ps = Ack (toAckInfo $ fromPeerPacketNumbers ps) 0+    mkAck ps =+        Ack (capAckInfo maxAckFrameSize $ toAckInfo $ fromPeerPacketNumbers ps) noDelay +-- | What an ACK frame may take.  A 1200-octet 1-RTT packet holding the+--   largest STREAM frame built here (1059) has about 100 octets left+--   after its header and tag; more would be cut off when encoded.+maxAckFrameSize :: Int+maxAckFrameSize = 64+ mkHeader :: Connection -> EncryptionLevel -> IO Header mkHeader conn lvl = do     ver <- getVersion conn@@ -332,6 +341,9 @@             then                 E.throwIO ExitConnection             else do+                -- Between two sends, so this thread is not inside a call+                -- on any of them.+                closeRetiredSockets conn                 sendP conn                 loop     logAction msg = connDebugLog conn ("debug: sender: " <> msg)@@ -350,7 +362,10 @@         SwStrm tx -> sendTxStreamData conn tx  sendFinal :: Connection -> IO ()-sendFinal conn = loop 30+sendFinal conn = do+    -- Draining: nothing goes to a peer that has closed.+    closed <- isPeerClosed conn+    unless closed $ loop 30   where     msg = "sendFinal " ++ if isServer conn then "Server" else "Client"     loop :: Int -> IO ()@@ -430,6 +445,10 @@     rs <- adjustForRetransmit conn xs     return (r : rs) adjustForRetransmit conn (Datagram{} : xs) = adjustForRetransmit conn xs+-- RFC 9000 Sec 13.3: "Responses to path validation using PATH_RESPONSE+-- frames are sent just once."  The peer sends a new PATH_CHALLENGE if+-- it needs one.  Resent, a PATH_RESPONSE flood would never leave us.+adjustForRetransmit conn (PathResponse{} : xs) = adjustForRetransmit conn xs adjustForRetransmit conn (x : xs) = do     rs <- adjustForRetransmit conn xs     return (x : rs)@@ -508,10 +527,15 @@  sendTxStreamData :: Connection -> TxStreamData -> IO () sendTxStreamData conn (TxStreamData s dats len fin0) = do-    fin <- packFin conn s fin0-    if len < limitation-        then sendStreamSmall conn s dats fin len-        else sendStreamLarge conn s dats fin+    reset <- isTxStreamReset s+    -- RFC 9000 Sec 3.3: no STREAM frame for a stream in "Reset Sent".+    if reset+        then when fin0 $ syncFinTx s+        else do+            fin <- packFin conn s fin0+            if len < limitation+                then sendStreamSmall conn s dats fin len+                else sendStreamLarge conn s dats fin  sendStreamSmall :: Connection -> Stream -> [StreamData] -> Bool -> Int -> IO () sendStreamSmall conn s0 dats0 fin0 len0 = do@@ -532,8 +556,19 @@         case mx of             Nothing -> do                 yield-                tryPeekSendStreamQ conn-            Just _ -> return mx+                tryPeekSendStreamQ conn >>= dropReset+            Just _ -> dropReset mx+    -- What a reset stream has queued goes nowhere; see 'sendTxStreamData'.+    dropReset mx = case mx of+        Just (TxStreamData s1 _ _ fin1) -> do+            reset <- isTxStreamReset s1+            if reset+                then do+                    _ <- takeSendStreamQ conn -- cf tryPeek+                    when fin1 $ syncFinTx s1+                    tryPeek+                else return mx+        Nothing -> return mx     loop         :: Stream         -> Frame@@ -582,6 +617,9 @@     sid = streamId s     loop [] = return ()     loop dats = do+        reset <- isTxStreamReset s+        unless reset $ send dats+    send dats = do         let (dats1, dats2) = splitChunks dats             len = totalLen dats1         off <- getTxStreamOffset s len

@@ -9,6 +9,8 @@     liveConnections,     spawnConnection,     waitNoConnection,+    spawnDispatcher,+    waitNoDispatcher,     tokenMgr,     genStatelessReset,     stopServerNow,@@ -22,7 +24,6 @@     ServerState (..), ) where -import Control.Concurrent import Control.Concurrent.STM import qualified Control.Monad.STM as STM import qualified Control.Exception as E@@ -31,7 +32,7 @@ import Data.Map.Strict (Map) import qualified Data.Map.Strict as M import Network.ByteOrder-import Network.Control (LRUCacheRef, Rate, getRate, newRate)+import Network.Control (LRUCacheRef, Rate, newRate) import qualified Network.Control as LRUCache import Network.Socket (     isClosedSocketError,@@ -39,7 +40,6 @@     Socket,  ) import qualified Network.Socket.ByteString as NSB-import System.Log.FastLogger import qualified System.ThreadManager as T  import Network.QUIC.Common@@ -51,7 +51,6 @@ import Network.QUIC.Logger import Network.QUIC.Packet import Network.QUIC.Parameters-import Network.QUIC.Qlog import Network.QUIC.Types  ----------------------------------------------------------------@@ -63,14 +62,16 @@     , genStatelessReset :: CID -> StatelessResetToken     , statelessResetRate :: Rate     , connectionThreads :: T.ThreadManager+    , dispatcherThreads :: T.ThreadManager+    -- ^ The threads reading the listening sockets.  They are waited for+    -- rather than killed: 'stopServerNow' is what ends them, and a thread+    -- blocked in a socket call is not reachable by an exception on Windows+    -- anyway.     , stopServerNow :: IO ()     -- ^ Stopping this server: what the shutdown handler is handed, and     -- what a connection's 'stop' reaches it by.     } -statelessResetLimit :: Int-statelessResetLimit = 20- newDispatch :: ServerConfig -> IO () -> IO Dispatch newDispatch ServerConfig{..} stopNow =     Dispatch@@ -80,6 +81,7 @@         <*> makeGenStatelessReset         <*> newRate         <*> T.newThreadManager+        <*> T.newThreadManager         <*> pure stopNow   where     conf =@@ -130,6 +132,16 @@ waitNoConnection :: Dispatch -> IO () waitNoConnection Dispatch{..} = T.waitUntilAllGone connectionThreads +-- | Running a dispatcher on a thread of its own.+spawnDispatcher :: Dispatch -> IO () -> IO ()+spawnDispatcher Dispatch{..} = T.forkManaged dispatcherThreads "QUIC dispatcher"++-- | Waiting until every dispatcher has returned.  Nothing is left reading+--   the listening sockets once this does, which is what lets the caller+--   close them.+waitNoDispatcher :: Dispatch -> IO ()+waitNoDispatcher Dispatch{..} = T.waitUntilAllGone dispatcherThreads+ ----------------------------------------------------------------  -- Source CID -> RecvQ@@ -173,8 +185,9 @@     -> TVar ServerState     -> (Accept -> IO ())     -> Socket-    -> IO ThreadId-runDispatcher d conf stvar forkConn mysock = forkIO $ dispatcher d conf stvar forkConn mysock+    -> IO ()+runDispatcher d conf stvar forkConn mysock =+    spawnDispatcher d $ dispatcher d conf stvar forkConn mysock  data ServerState = Running | Stopped deriving (Eq, Show) @@ -227,7 +240,8 @@     logAction _msg = return ()      -- One datagram's failure is one datagram's problem here too: a receive-    -- which fails is tried again rather than ending the dispatcher.+    -- which fails is tried again, however often, rather than ending the+    -- dispatcher.     safeRecv rcv = do         ex <- E.try rcv         case ex of@@ -239,7 +253,7 @@                 Just e | isClosedSocketError e -> E.throwIO se                 _ -> do                     logAction $ "recv again: " <> bhow se-                    rcv+                    safeRecv rcv  ---------------------------------------------------------------- @@ -460,43 +474,25 @@     (cpkt@(CryptPacket (Short dCID) _), lvl, siz) = do         mconn <- lookupConnectionDict dstTable dCID         case mconn of-            Nothing -> do-                -- Three times rule for stateless reset-                -- Our packet size is 1280-                when (bytes > 427) $ do-                    srRate <- getRate statelessResetRate-                    -- fixme: hard coding-                    when (srRate < statelessResetLimit) $ do-                        statelessReset <- makeStatelessReset $ genStatelessReset dCID+            -- The size first, so that a packet too small to answer uses+            -- up nothing of the rate.+            Nothing -> case statelessResetSizeFor bytes of+                Nothing -> return ()+                Just rsiz -> do+                    ok <- withinStatelessResetLimit statelessResetRate+                    when ok $ do+                        statelessReset <- makeStatelessReset rsiz $ genStatelessReset dCID                         send' statelessReset                         logAction $ "Stateless reset is sent to " <> bhow peersa             Just conn -> do                 alive <- getAlive conn                 when alive $ do                     void $ setSocket conn mysock -- fixme-                    curCID <- getMyCID conn-                    -- setMyCID is not called here since setMyCID is-                    -- done in Receiver.-                    let cidChanged = curCID /= dCID-                    mPathInfo <- findPathInfo conn peersa-                    case mPathInfo of-                        Nothing -> do-                            -- Taken here, before the thread rather than-                            -- inside it, so that a peer sending from an-                            -- address this connection has not seen gets one-                            -- thread for the migration and not one for every-                            -- packet.  'findPathInfo' knows the address in-                            -- use and the one before it and nothing else, so-                            -- a sender that moves its port -- which asks no-                            -- spoofing of it -- is unknown every time, and-                            -- this runs on packets nothing has authenticated-                            -- yet.-                            claimed <- claimMigration conn-                            when claimed $-                                forkManaged conn "QUIC migration" $-                                    onClientMigration conn dCID peersa cidChanged-                        _ -> return ()-                    writeRecvQ (connRecvQ conn) $ mkReceivedPacket cpkt tim siz lvl+                    -- Whether the peer has moved is for the receiver to say,+                    -- once the packet has decrypted: anyone who has seen the+                    -- CID can send one from anywhere.+                    let rpkt = mkReceivedPacket cpkt tim siz lvl+                    writeRecvQ (connRecvQ conn) rpkt{rpPeerAddr = Just peersa} ---------------------------------------------------------------- dispatch     Dispatch{..}@@ -516,29 +512,10 @@             Just conn -> do                 -- fixme: is this block necessary?                 void $ setSocket conn mysock-                mPathInfo <- findPathInfo conn peersa-                when (isNothing mPathInfo) $ do-                    pathInfo <- newPathInfo peersa-                    addPathInfo conn pathInfo+                -- No new path from here: this packet has not decrypted, and+                -- the peer may not migrate during the handshake anyway+                -- (RFC 9000 Sec 9).                 writeRecvQ (connRecvQ conn) $ mkReceivedPacket cpkt tim siz lvl  recvServer :: RecvQ -> IO ReceivedPacket recvServer = readRecvQ---- | Carry out the migration the caller has claimed with 'claimMigration'.-onClientMigration :: Connection -> CID -> SockAddr -> Bool -> IO ()-onClientMigration conn newdCID peersa cidChanged = handleLogUnit logAction $ do-    -- fixme: should not block-    mcidinfo <--        if cidChanged-            then timeout (Microseconds 100000) "onClientMigration" $ waitPeerCID conn-            else return Nothing -- PathChallenge only, no RetireConnectionID-    let msg = "Migration: " <> bhow peersa <> " (" <> bhow newdCID <> ")"-    qlogDebug conn $ Debug $ toLogStr msg-    connDebugLog conn $ "debug: onClientMigration: " <> msg-    pathInfo <- newPathInfo peersa-    -- assumed that this PathInfo is not stored in PeerInfo-    addPathInfo conn pathInfo-    validatePath conn pathInfo mcidinfo-  where-    logAction msg = connDebugLog conn ("debug: onClientMigration: " <> msg)

@@ -8,7 +8,6 @@     stop, ) where -import Control.Concurrent import Control.Concurrent.Async import Control.Concurrent.STM import qualified Control.Exception as E@@ -52,7 +51,7 @@     -- onServerReady, and left anyone waiting on that hook waiting for good.     -- An IOSpec run wedged for two and a half days that way, on a port the     -- previous test had not finished releasing.-    E.bracket (setup stvar stopNow) (teardown stopNow) $ \(_, _, _) -> handleLogUnit debugLog $ do+    E.bracket (setup stvar stopNow) (teardown stopNow) $ \_ -> handleLogUnit debugLog $ do         onServerReady $ scHooks conf         atomically $ do             st <- readTVar stvar@@ -71,29 +70,35 @@                     runServer conf server dispatch acc         flip E.onException (clearDispatch dispatch) $ do             ssas <- openAll $ scAddresses conf-            tids <--                runAll dispatch conf stvar forkConn ssas-                    `E.onException` mapM_ NS.close ssas-            return (dispatch, tids, ssas)-    -- Telling the dispatchers first.  They end where they wait, on the stop-    -- variable 'stopping' sets, and the 'killThread' below then has nothing-    -- to reach -- which matters because on Windows it could not have-    -- reached them anyway: a thread blocked in a socket call there is not-    -- interruptible.+            runAll dispatch conf stvar forkConn ssas+                `E.onException` do+                    stopNow+                    waitNoDispatcher dispatch+                    mapM_ NS.unsafeClose ssas+            return (dispatch, ssas)+    -- Telling the dispatchers, and then waiting for them.  They end where+    -- they wait, on the stop variable 'stopping' sets; nothing is thrown at+    -- them, which matters because on Windows an exception would not reach a+    -- thread blocked in a socket call anyway.     --+    -- The waiting is what lets the sockets be closed below: a dispatcher+    -- which has returned has given up its receive, and the connections+    -- waited for after it are what send on those sockets.  Telling a thread+    -- to stop is not the same as it having stopped.+    --     -- The usual way here is through the shutdown handler, which has already     -- done this; doing it again is free.  The way that had not, until now,     -- is 'run' being cancelled from outside.-    teardown :: IO () -> (Dispatch, [ThreadId], [NS.Socket]) -> IO ()-    teardown stopNow (dispatch, tids, ssas) = do+    teardown :: IO () -> (Dispatch, [NS.Socket]) -> IO ()+    teardown stopNow (dispatch, ssas) = do         stopNow         clearDispatch dispatch-        mapM_ killThread tids+        waitNoDispatcher dispatch         shutdownConnections conf dispatch-        mapM_ NS.close ssas+        mapM_ NS.unsafeClose ssas     openAll [] = return []     openAll (a : as) =-        E.bracketOnError (serverSocket a) NS.close $ \s -> (s :) <$> openAll as+        E.bracketOnError (serverSocket a) NS.unsafeClose $ \s -> (s :) <$> openAll as  -- | Running a QUIC server. --   The action is executed with a new connection@@ -105,7 +110,7 @@     let stopNow = stopping stvar     installShutdownHandler conf stopNow     -- As in 'run'.-    E.bracket (setup stvar stopNow) (teardown stopNow) $ \(_, _) -> handleLogUnit debugLog $ do+    E.bracket (setup stvar stopNow) (teardown stopNow) $ \_ -> handleLogUnit debugLog $ do         onServerReady $ scHooks conf         atomically $ do             st <- readTVar stvar@@ -120,14 +125,17 @@                 spawnConnection dispatch $                     runServer conf server dispatch acc         flip E.onException (clearDispatch dispatch) $ do-            tids <- runAll dispatch conf stvar forkConn ssas-            return (dispatch, tids)-    -- As in 'run'.-    teardown :: IO () -> (Dispatch, [ThreadId]) -> IO ()-    teardown stopNow (dispatch, tids) = do+            runAll dispatch conf stvar forkConn ssas+                `E.onException` (stopNow >> waitNoDispatcher dispatch)+            return dispatch+    -- As in 'run'.  The caller's sockets are not closed here, but the+    -- dispatchers are still waited for: the caller is free to close them+    -- the moment this returns.+    teardown :: IO () -> Dispatch -> IO ()+    teardown stopNow dispatch = do         stopNow         clearDispatch dispatch-        mapM_ killThread tids+        waitNoDispatcher dispatch         shutdownConnections conf dispatch  -- | Handing the caller the action that stops this server.@@ -175,10 +183,6 @@ -- Hearing the peers is another matter: once the successor has the port, on -- Linux it is the successor that receives.  See Note [Binding the same port -- twice].  Sending still works, which is what this relies on.------ Each is ended from a thread of its own, because 'throwTo' waits for the--- exception to be taken and a connection whose application is slow to--- unwind would hold up the rest.  -- | Telling every connection to stop, and waiting until none is left. --@@ -189,12 +193,9 @@ -- peer heard nothing and sat until its idle timeout.  It is also a thread -- reading a descriptor number that the kernel may hand to someone else. ----- Each connection is told once.  Telling it again a second later cannot--- hurry a connection that is already closing down under 'mask', and what it--- does reach is the CONNECTION_CLOSE being sent -- which it then interrupts--- and loses, which is the thing this is here to prevent.  The second pass--- is only to tell connections that were not yet in the table when the first--- one looked.+-- Telling is 'abortConnection', which only asks, so telling a connection+-- again does nothing.  Every second the table is looked at again, for the+-- connections that were not in it yet. -- -- There is no cap on the wait.  An application that never finishes with a -- connection therefore keeps the server from returning, which is a@@ -202,32 +203,28 @@ -- what this changed.  http3 caps the wait it does of its own, so an H3 -- connection comes back within that cap however its handler behaves. shutdownConnections :: ServerConfig -> Dispatch -> IO ()-shutdownConnections conf dispatch = loop []+shutdownConnections conf dispatch = loop   where     (err, reason) = scCloseReason conf     shut conn = abortConnection conn err reason-    loop told = do-        conns <- liveConnections dispatch-        let unTold = [conn | conn <- conns, mainThreadId conn `notElem` told]-        mapM_ (void . forkIO . shut) unTold+    loop = do+        liveConnections dispatch >>= mapM_ shut         done <- T.timeout 1000000 $ waitNoConnection dispatch         case done of             Just () -> return ()-            Nothing -> loop $ map mainThreadId unTold ++ told+            Nothing -> loop --- | Running a dispatcher on each socket, killing the ones already running if---   a later one cannot be started.+-- | Running a dispatcher on each socket.  The caller ends them all+--   together, by telling them, so there is nothing to undo here for the+--   ones already running. runAll     :: Dispatch     -> ServerConfig     -> TVar ServerState     -> (Accept -> IO ())     -> [NS.Socket]-    -> IO [ThreadId]-runAll _ _ _ _ [] = return []-runAll dispatch conf stvar forkConn (s : ss) =-    E.bracketOnError (runDispatcher dispatch conf stvar forkConn s) killThread $-        \t -> (t :) <$> runAll dispatch conf stvar forkConn ss+    -> IO ()+runAll dispatch conf stvar forkConn = mapM_ $ runDispatcher dispatch conf stvar forkConn  -- Typically, ConnectionIsClosed breaks acceptStream. -- And the exception should be ignored.@@ -282,6 +279,8 @@                 c2 = labelMe "concurrently2" >> concurrently_ c1 s3                 c3 = labelMe "concurrently3" >> concurrently_ c2 s4                 c4 = labelMe "concurrently4" >> concurrently_ c3 s5+                c4' = labelMe "concurrently4'" >> concurrently_ c4 s7+                s7 = labelMe "abort watcher" >> watchAbortRequest conn                 -- Telling the peer here, and not below, because here is the                 -- one place where it can be done at all: the nested                 -- 'concurrently_'s above have waited for every protocol@@ -309,24 +308,24 @@                 c5 =                     labelMe "concurrently5"                         >> concurrently_-                            (tellThenRaise (c4 `E.catch` \(_ :: InternalControl) -> return ()))+                            (tellThenRaise (c4' `E.catch` \(_ :: InternalControl) -> return ()))                             s6                 runThreads = c5-            ex <- E.try runThreads+            ex <- E.try runThreads >>= abortRequested conn             sendFinal conn             setConnectionClosed conn             closure conn ldcc ex   where     open = createServerConnection conf dispatch acc-    -- Stop this connection's own threads and see them out before the-    -- buffers they use are freed.  Nothing stopped them here at all: a path-    -- validation outlived the connection it was validating a path for, and-    -- went on reading and writing buffers that had been handed back.+    -- 'killReaders' stops this connection's own threads and sees them out,+    -- which has to happen before the buffers they use are freed.  Nothing+    -- stopped them here at all: a path validation outlived the connection it+    -- was validating a path for, and went on reading and writing buffers+    -- that had been handed back.     clse connRes = do         let conn = connResConnection connRes         setDead conn         killReaders conn-        waitManaged conn         freeResources conn     -- Say why the connection ended.  This used to discard it, so a server     -- connection that died of anything 'closure' does not turn into a

@@ -7,12 +7,17 @@ ) where  import qualified Control.Exception as E+import Control.Monad (when) import Data.IP (IP, toSockAddr) import qualified Data.List.NonEmpty as NE import Network.Socket -natRebinding :: SockAddr -> IO Socket-natRebinding sa = E.bracketOnError open close return+-- | A new socket for the peer, connected to it if asked.  It is closed+--   here if the connect fails, since nothing else holds it yet.+natRebinding :: Bool -> SockAddr -> IO Socket+natRebinding connected sa = E.bracketOnError open unsafeClose $ \s -> do+    when connected $ connect s sa+    return s   where     family = sockAddrFamily sa     open = socket family Datagram defaultProtocol@@ -25,7 +30,7 @@ clientSocket :: HostName -> ServiceName -> IO (Socket, SockAddr) clientSocket host port = do     addr <- NE.head <$> getAddrInfo (Just hints) (Just host) (Just port)-    E.bracketOnError (openSocket addr) close $ \s -> return (s, addrAddress addr)+    E.bracketOnError (openSocket addr) unsafeClose $ \s -> return (s, addrAddress addr)   where     hints = defaultHints{addrSocketType = Datagram, addrFlags = [AI_ADDRCONFIG]} @@ -53,7 +58,7 @@ -- processes at random.  serverSocket :: (IP, PortNumber) -> IO Socket-serverSocket ip = E.bracketOnError open close $ \s -> do+serverSocket ip = E.bracketOnError open unsafeClose $ \s -> do     setSocketOption s ReuseAddr 1 #if defined(darwin_HOST_OS) || defined(freebsd_HOST_OS) || defined(netbsd_HOST_OS) || defined(openbsd_HOST_OS)     setSocketOption s ReusePort 1

@@ -9,12 +9,14 @@     RecvStreamQ (..),     RxStreamData (..),     RxBounds (..),+    Departed (..),     Length,     syncFinTx,     waitFinTx,      -- * Misc-    getTxStreamFinalSize,+    resetTxStream,+    isTxStreamReset,     getTxStreamOffset,     isTxStreamClosed,     setTxStreamClosed,@@ -27,10 +29,13 @@     FinalSizeProblem (..),     getRxBounds,     noteRxFrame,+    noteRxFrameSTM,     noteRxFinalSize,-    addRxCounted,-    takeRxUncounted,-    takeRxUnread,+    getRxBoundsSTM,+    addRxCredited,+    addRxCreditedSTM,+    takeRxOwedSTM,+    getRxConsumed,     readStreamFlowTx,     addTxStreamData,     setTxMaxStreamData,

@@ -1,41 +0,0 @@-module Network.QUIC.Stream.Frag where--import Data.Sequence (-    Seq,-    ViewL (..),-    ViewR (..),-    dropWhileL,-    viewl,-    viewr,-    (<|),- )--class Frag a where-    currOff :: a -> Int-    nextOff :: a -> Int-    shrink :: Int -> a -> a--instance Frag a => Frag (Seq a) where-    currOff s = case viewl s of-        EmptyL -> error "Seq is empty (1)"-        x :< _ -> currOff x-    nextOff s = case viewr s of-        EmptyR -> error "Seq is empty (2)"-        _ :> x -> nextOff x-    shrink = shrinkSeq--shrinkSeq :: Frag a => Int -> Seq a -> Seq a-shrinkSeq n s0 = case viewl s of-    EmptyL -> error "shrinkSeq"-    x :< xs-        | nextOff x == n -> xs-        | otherwise -> shrink n x <| xs-  where-    s = dropWhileL (\y -> not (currOff y <= n && n <= nextOff y)) s0--data F = F Int Int deriving (Show)--instance Frag F where-    currOff (F s _) = s-    nextOff (F _ e) = e-    shrink n (F s e) = if s <= n && n <= e then F n e else error "shrink"

@@ -2,7 +2,8 @@ {-# LANGUAGE RecordWildCards #-}  module Network.QUIC.Stream.Misc (-    getTxStreamFinalSize,+    resetTxStream,+    isTxStreamReset,     getTxStreamOffset,     isTxStreamClosed,     setTxStreamClosed,@@ -15,10 +16,13 @@     FinalSizeProblem (..),     getRxBounds,     noteRxFrame,+    noteRxFrameSTM,     noteRxFinalSize,-    addRxCounted,-    takeRxUncounted,-    takeRxUnread,+    getRxBoundsSTM,+    addRxCredited,+    addRxCreditedSTM,+    takeRxOwedSTM,+    getRxConsumed,     --     readStreamFlowTx,     addTxStreamData,@@ -39,9 +43,19 @@  ---------------------------------------------------------------- -getTxStreamFinalSize :: Stream -> IO Offset-getTxStreamFinalSize Stream{..} = streamOffset <$> readIORef streamStateTx+-- | Reset the sending part and return its final size: every octet handed+--   to the sender, sent or still queued.  The sender drops what is queued,+--   so nothing goes out past it.+resetTxStream :: Stream -> IO Offset+resetTxStream s@Stream{..} = do+    -- Before the read: what is queued after it is dropped, not counted.+    atomicWriteIORef streamResetTx True+    setTxStreamClosed s+    txfSent <$> readTVarIO streamFlowTx +isTxStreamReset :: Stream -> IO Bool+isTxStreamReset Stream{..} = readIORef streamResetTx+ getTxStreamOffset :: Stream -> Int -> IO Offset getTxStreamOffset Stream{..} len = atomicModifyIORef' streamStateTx get   where@@ -153,75 +167,71 @@       DataPastFinalSize     deriving (Eq, Show) --- | Taking in where one STREAM frame says the stream reaches, and whether it---   ends it.  Nothing is counted here: a frame may still turn out to be a---   duplicate, and only what is taken counts. -- | What the receiving side has seen of where this stream ends. getRxBounds :: Stream -> IO RxBounds-getRxBounds Stream{..} = readIORef streamRxBounds+getRxBounds Stream{..} = readTVarIO streamRxBounds -noteRxFrame :: Stream -> Int -> Bool -> IO (Maybe FinalSizeProblem)-noteRxFrame Stream{..} end fin = atomicModifyIORef' streamRxBounds note+-- | Answering with how far the frame takes the stream past what has been+--   seen of it: what the connection's flow controller is to count as+--   received for it (RFC 9000 Sec 4.1).  A copy of what was seen is 0.+noteRxFrame :: Stream -> Int -> Bool -> IO (Either FinalSizeProblem Int)+noteRxFrame s end fin = atomically $ noteRxFrameSTM s end fin++-- | 'noteRxFrame' as part of a transaction: see+--   'Network.QUIC.Connection.StreamTable.noteRxArrival'.+noteRxFrameSTM :: Stream -> Int -> Bool -> STM (Either FinalSizeProblem Int)+noteRxFrameSTM Stream{..} end fin = stateTVar streamRxBounds $ \b -> note b   where     note b@RxBounds{..}         | fin = case rxFinal of             Just f-                | f /= end -> (b, Just FinalSizeChanged)+                | f /= end -> (Left FinalSizeChanged, b)             _-                | end < rxHighest -> (b, Just FinalSizeTooSmall)-                | otherwise ->-                    (b{rxFinal = Just end, rxHighest = max rxHighest end}, Nothing)+                | end < rxHighest -> (Left FinalSizeTooSmall, b)+                | otherwise -> (Right (end - rxHighest), b{rxFinal = Just end, rxHighest = end})         | otherwise = case rxFinal of             Just f-                | end > f -> (b, Just DataPastFinalSize)-            _ -> (b{rxHighest = max rxHighest end}, Nothing)+                | end > f -> (Left DataPastFinalSize, b)+            _ -> (Right (max 0 (end - rxHighest)), b{rxHighest = max rxHighest end})  -- | The same for the final size a RESET_STREAM carries.-noteRxFinalSize :: Stream -> Int -> IO (Maybe FinalSizeProblem)+noteRxFinalSize :: Stream -> Int -> IO (Either FinalSizeProblem Int) noteRxFinalSize s end = noteRxFrame s end True --- | Counting octets the connection's flow controller has taken for the---   stream.-addRxCounted :: Stream -> Int -> IO ()-addRxCounted Stream{..} n =-    atomicModifyIORef'' streamRxBounds $ \b -> b{rxCounted = rxCounted b + n}+-- | 'getRxBounds' as part of a transaction.+getRxBoundsSTM :: Stream -> STM RxBounds+getRxBoundsSTM Stream{..} = readTVar streamRxBounds --- | The octets the peer spent on the stream that will never arrive, and---   which the connection's flow controller has therefore not counted.------ RFC 9000 Sec 4.5: "A receiver SHOULD use the final size to account for all--- bytes sent on the stream in its connection-level flow controller."  Left--- uncounted, the window we advertise falls behind what the peer believes it--- has spent, by the tail of every stream it resets, until it has none left.------ Answered once: whatever is asked for here is counted from then on.-takeRxUncounted :: Stream -> IO Int-takeRxUncounted Stream{..} = atomicModifyIORef' streamRxBounds take'-  where-    take' b@RxBounds{..} = case rxFinal of-        Just f-            | f > rxCounted ->-                (b{rxCounted = f, rxCredited = rxCredited + f - rxCounted}, f - rxCounted)-        _ -> (b, 0)+-- | Counting octets as consumed on the stream's behalf though the+--   application has not read them.+addRxCredited :: Stream -> Int -> IO ()+addRxCredited s n = atomically $ addRxCreditedSTM s n --- | The octets of the stream the connection's flow controller has counted---   and the application will never read.+addRxCreditedSTM :: Stream -> Int -> STM ()+addRxCreditedSTM Stream{..} n =+    modifyTVar' streamRxBounds $ \b -> b{rxCredited = rxCredited b + n}++-- | What the connection's flow controller has counted as received for the+--   stream and nobody has counted as consumed, given what the application+--   has consumed: what is owed back once the application is done with it. ----- What it read it has already counted itself, in 'recvStream'.  What it--- leaves -- a request whose body a server answers without reading, say ----- was counted as received and is never counted as consumed, so the window we--- advertise stays that much smaller for the rest of the connection.  The peer--- is then blocked by octets nobody is waiting for.+-- What it read it has already counted itself, in 'recvStream'.  The rest --+-- a request body a server answers without reading, or what was lost on the+-- way of a stream whose end is known -- was spent by the peer all the same,+-- and left uncounted the window we advertise stays that much smaller for+-- the rest of the connection. -- -- Answered once for each octet: what is answered here is counted as -- credited.-takeRxUnread :: Stream -> IO Int-takeRxUnread Stream{..} = do-    consumed <- rxfConsumed <$> readIORef streamFlowRx-    atomicModifyIORef' streamRxBounds $ take' consumed+takeRxOwedSTM :: Stream -> Int -> STM Int+takeRxOwedSTM Stream{..} consumed = stateTVar streamRxBounds take'   where-    take' consumed b@RxBounds{..}-        | owed > 0 = (b{rxCredited = rxCredited + owed}, owed)-        | otherwise = (b, 0)+    take' b@RxBounds{..}+        | owed > 0 = (owed, b{rxCredited = rxCredited + owed})+        | otherwise = (0, b)       where-        owed = rxCounted - consumed - rxCredited+        owed = rxHighest - consumed - rxCredited++-- | What the application has read of the stream, as its window counts it.+getRxConsumed :: Stream -> IO Int+getRxConsumed Stream{..} = rxfConsumed <$> readIORef streamFlowRx

@@ -5,13 +5,16 @@ import Control.Concurrent.STM import Data.ByteString (ByteString) +import {-# SOURCE #-} Network.QUIC.Connection.Types (peerClosedSTM) import Network.QUIC.Stream.Types  putRecvStreamQ :: Stream -> ByteString -> IO () putRecvStreamQ Stream{..} = atomically . writeTQueue (recvStreamQ streamRecvQ)  takeRecvStreamQ :: Stream -> IO ByteString-takeRecvStreamQ Stream{..} = atomically $ readTQueue $ recvStreamQ streamRecvQ+takeRecvStreamQ Stream{..} =+    atomically $+        readTQueue (recvStreamQ streamRecvQ) `orElse` peerClosedSTM streamConnection  tryTakeRecvStreamQ :: Stream -> IO (Maybe ByteString) tryTakeRecvStreamQ Stream{..} = atomically $ tryReadTQueue $ recvStreamQ streamRecvQ

@@ -1,3 +1,4 @@+{-# LANGUAGE MultiWayIf #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-} @@ -10,14 +11,12 @@ ) where  import qualified Data.ByteString as BS-import Data.Sequence (Seq)-import qualified Data.Sequence as Seq+import Data.Map.Strict (Map)+import qualified Data.Map.Strict as Map  import Network.QUIC.Imports-import Network.QUIC.Stream.Frag import Network.QUIC.Stream.Misc import Network.QUIC.Stream.Queue-import qualified Network.QUIC.Stream.Skew as Skew import Network.QUIC.Stream.Types import Network.QUIC.Types @@ -99,8 +98,10 @@       OverLimit     | -- | Past the number of separate pieces we will hold for one stream.       TooFragmented-    | Duplicated-    | Reassembled+    | -- | Nothing that had not been delivered or held already.+      Duplicated+    | -- | Taken, with the number of octets which are new.+      Reassembled Int     deriving (Eq, Show)  putRxStreamData :: Stream -> RxStreamData -> IO FlowCntl@@ -134,80 +135,58 @@         then return OverLimit         else tryReassemble s rx put (return ()) --- fin of StreamState off fin means see-fin-already.+-- | Take the octets of a fragment which are neither delivered nor held,+--   deliver whatever that makes contiguous, and say how many were new.+--+-- What is held is kept disjoint, so a copy sent again with other boundaries+-- -- a retransmission, or a CRYPTO flight repacketized after a PTO -- adds+-- exactly what it brings, and the count of pieces held is the true one. tryReassemble     :: Stream -> RxStreamData -> (StreamData -> IO ()) -> IO () -> IO FlowCntl-tryReassemble Stream{} (RxStreamData "" _ _ False) _ _ = return Duplicated-tryReassemble Stream{..} x@(RxStreamData "" off _ True) _ putFin = do-    si0@(StreamState off0 fin0) <- readIORef streamStateRx-    let si1 = si0{streamFin = True}-    if fin0-        then do-            -- stdoutLogger "Illegal Fin" -- fixme-            return Duplicated-        else case off `compare` off0 of-            LT -> return Duplicated-            EQ -> do-                writeIORef streamStateRx si1-                putFin-                return Reassembled-            GT -> do-                writeIORef streamStateRx si1-                hold streamReass x-tryReassemble Stream{..} x@(RxStreamData dat off len False) put putFin = do-    si0@(StreamState off0 _) <- readIORef streamStateRx-    case off `compare` off0 of-        LT -> return Duplicated-        EQ -> do-            put dat-            loop si0 (off0 + len)-            return Reassembled-        GT -> hold streamReass x-  where-    loop si0 xff = do-        mrxs <- atomicModifyIORef' streamReass $ \(n, sk) ->-            let (sk', mrxs) = Skew.deleteMinIf xff sk-             in ((n - maybe 0 length mrxs, sk'), mrxs)-        case mrxs of-            Nothing -> writeIORef streamStateRx si0{streamOffset = xff}-            Just rxs -> do-                mapM_ (put . rxstrmData) rxs-                let xff1 = nextOff rxs-                if hasFin rxs-                    then do-                        putFin-                    else do-                        loop si0 xff1-tryReassemble Stream{..} x@(RxStreamData dat off len True) put putFin = do-    si0@(StreamState off0 fin0) <- readIORef streamStateRx-    let si1 = si0{streamFin = True}-    if fin0-        then return Duplicated-        else case off `compare` off0 of-            LT -> return Duplicated-            EQ -> do-                let off1 = off0 + len-                writeIORef streamStateRx si1{streamOffset = off1}-                put dat-                putFin-                return Reassembled-            GT -> do-                writeIORef streamStateRx si1-                hold streamReass x+tryReassemble Stream{..} (RxStreamData dat off len fin) put putFin = do+    StreamState off0 fin0 <- readIORef streamStateRx+    Reass held mfin <- readIORef streamReass+    let new = uncovered held (max off off0) $ BS.drop (off0 - off) dat+        newLen = sum $ map (BS.length . snd) new+        -- 'streamFin' is set by the first FIN, or when we stop receiving.+        finNew = fin && not fin0+        mfin1 = if finNew then Just (off + len) else mfin+        (ready, held1, off1) = deliverable off0 $ foldr (uncurry Map.insert) held new+        done = mfin1 == Just off1+    if+        | newLen == 0 && not finNew -> return Duplicated+        | Map.size held1 > maxReassFragments -> return TooFragmented+        | otherwise -> do+            writeIORef streamReass $ Reass held1 $ if done then Nothing else mfin1+            atomicModifyIORef'' streamStateRx $ \(StreamState _ f) ->+                StreamState off1 (f || fin)+            mapM_ put ready+            when done putFin+            return $ Reassembled newLen --- | Keep a fragment that cannot be delivered yet, unless we are already---   holding as many as we are willing to.------ Flow control bounds the octets, not the pieces, and a peer that sends its--- window one octet at a time at scattered offsets pays for the octets while--- we pay for the pieces.-hold :: IORef (Int, Skew.Skew RxStreamData) -> RxStreamData -> IO FlowCntl-hold ref x = atomicModifyIORef' ref $ \st@(n, sk) ->-    if n >= maxReassFragments-        then (st, TooFragmented)-        else ((n + 1, Skew.insert x sk), Reassembled)+-- | The parts of the octets at an offset which no held piece covers.+uncovered+    :: Map Offset StreamData -> Offset -> StreamData -> [(Offset, StreamData)]+uncovered held = go+  where+    go s d+        | BS.null d = []+        | Just (k, v) <- Map.lookupLE s held+        , k + BS.length v > s =+            let e = k + BS.length v in go e $ BS.drop (e - s) d+        | Just (k, _) <- Map.lookupGT s held+        , k < s + BS.length d =+            let (d1, d2) = BS.splitAt (k - s) d in (s, d1) : go k d2+        | otherwise = [(s, d)] -hasFin :: Seq RxStreamData -> Bool-hasFin s = case Seq.viewr s of-    Seq.EmptyR -> False-    _ Seq.:> x -> rxstrmFin x+-- | The pieces which follow on from an offset, without a gap.+deliverable+    :: Offset+    -> Map Offset StreamData+    -> ([StreamData], Map Offset StreamData, Offset)+deliverable o m = case Map.minViewWithKey m of+    Just ((k, d), m')+        | k == o ->+            let (ds, m'', o') = deliverable (o + BS.length d) m'+             in (d : ds, m'', o')+    _ -> ([], m, o)

@@ -1,91 +0,0 @@-module Network.QUIC.Stream.Skew (-    Skew (..),-    empty,-    insert,-    deleteMin,-    deleteMinIf,-    --  , deleteMin'-    --  , showSkew-) where--import Data.Maybe-import Data.Sequence (Seq, (><))-import qualified Data.Sequence as Seq-import Prelude hiding (minimum)--import Network.QUIC.Stream.Frag--------------------------------------------------------------------data Skew a = Leaf | Node (Skew a) (Seq a) (Skew a) deriving (Show)--empty :: Skew a-empty = Leaf---------------------------------------------------------------------- | Insertion. Worst-case: O(N), amortized: O(log N).-insert :: Frag a => a -> Skew a -> Skew a-insert x t = merge (Node Leaf (Seq.singleton x) Leaf) t---------------------------------------------------------------------- | Finding the minimum element. Worst-case: O(1).-minimum :: Skew a -> Maybe (Seq a)-minimum Leaf = Nothing-minimum (Node _ f _) = Just f---------------------------------------------------------------------- | Deleting the minimum element. Worst-case: O(N), amortized: O(log N).-deleteMin' :: Frag a => Skew a -> Skew a-deleteMin' Leaf = Leaf-deleteMin' (Node l _ r) = merge l r--deleteMin :: Frag a => Skew a -> (Skew a, Maybe (Seq a))-deleteMin h = (deleteMin' h, minimum h)--deleteMinIf :: Frag a => Int -> Skew a -> (Skew a, Maybe (Seq a))-deleteMinIf off h = case minimum h of-    jf@(Just f)-        | currOff f == off -> (deleteMin' h, jf)-        | currOff f < off && off < nextOff f -> (deleteMin' h, shrink off <$> jf)-        | nextOff f <= off -> (deleteMin' h, Nothing)-    _ -> (h, Nothing)---------------------------------------------------------------------- | Merging two heaps. Worst-case: O(N), amortized: O(log N).-merge :: Frag a => Skew a -> Skew a -> Skew a-merge t1 Leaf = t1-merge Leaf t2 = t2-merge t1@(Node l1 f1 r1) t2@(Node l2 f2 r2)-    | e1 < s2 = Node r1 f1 (merge l1 t2)-    | e2 < s1 = Node r2 f2 (merge l2 t1)-    | otherwise =-        let f12-                | e1 == s2 = f1 >< f2-                | s1 == e2 = f2 >< f1-                | s1 <= s2 && e2 <= e1 = f1-                | s2 <= s1 && e1 <= e2 = f2-                | s1 <= s2 = f1 >< shrink e1 f2-                | otherwise = f2 >< shrink e2 f1-         in Node (merge l1 l2) f12 (merge r1 r2)-  where-    s1 = currOff f1-    e1 = nextOff f1-    s2 = currOff f2-    e2 = nextOff f2--{--showSkew :: Show a => Skew a -> String-showSkew = showSkew' ""--showSkew' :: Show a => String -> Skew a -> String-showSkew' _    Leaf = "\n"-showSkew' pref (Node l x r) = show x ++ "\n"-                           ++ pref ++ "+ " ++ showSkew' pref' l-                           ++ pref ++ "+ " ++ showSkew' pref' r-  where-    pref' = "  " ++ pref--}

@@ -7,23 +7,22 @@     StreamState (..),     RxBounds (..),     emptyRxBounds,+    Departed (..),     RecvStreamQ (..),     RxStreamData (..),+    Reass (..),     Length,     syncFinTx,     waitFinTx, ) where -import Control.Concurrent import Control.Concurrent.STM-import qualified Data.ByteString as BS+import Data.Map.Strict (Map)+import qualified Data.Map.Strict as Map import Network.Control -import {-# SOURCE #-} Network.QUIC.Connection.Types+import {-# SOURCE #-} Network.QUIC.Connection.Types (Connection, peerClosedSTM) import Network.QUIC.Imports-import Network.QUIC.Stream.Frag-import Network.QUIC.Stream.Skew-import qualified Network.QUIC.Stream.Skew as Skew import Network.QUIC.Types  ----------------------------------------------------------------@@ -40,35 +39,45 @@     , streamStateTx :: IORef StreamState -- offset, fin     , streamStateRx :: IORef StreamState -- offset, fin     , streamRecvQ :: RecvStreamQ -- input bytestring-    , -- input stream fragments to streamQ, and how many of them-      streamReass :: IORef (Int, Skew RxStreamData)-    , streamSyncFinTx :: MVar ()+    , streamReass :: IORef Reass+    -- ^ Fragments which cannot be delivered yet+    , streamSyncFinTx :: TMVar ()+    , streamResetTx :: IORef Bool+    -- ^ Whether our sending part has been reset     , streamResetRx :: IORef (Maybe ApplicationProtocolError)     -- ^ The error code of a RESET_STREAM from the peer     , streamReleased :: IORef Bool     -- ^ Whether we are done with it and have counted it so-    , streamRxBounds :: IORef RxBounds+    , streamRxBounds :: TVar RxBounds     -- ^ What has been seen of the stream's end (RFC 9000 Sec 4.5)     }  -- | What the receiving side has seen of where a stream ends. data RxBounds = RxBounds-    { rxCounted :: Int-    -- ^ Octets of this stream the connection's flow controller has counted:-    --   every frame taken for it, in order or not-    , rxHighest :: Int-    -- ^ The largest offset plus length seen for it+    { rxHighest :: Int+    -- ^ The largest offset plus length seen for it.  The connection's flow+    --   controller counts this as received (RFC 9000 Sec 4.1), gaps and all.     , rxFinal :: Maybe Int     -- ^ Its final size, once that is known     , rxCredited :: Int-    -- ^ Octets counted against the connection's window on the stream's-    --   behalf, over and above what the application has read+    -- ^ Octets counted as consumed on the stream's behalf, over and above+    --   what the application has read     }     deriving (Eq, Show)  emptyRxBounds :: RxBounds-emptyRxBounds = RxBounds 0 0 Nothing 0+emptyRxBounds = RxBounds 0 Nothing 0 +-- | What is kept of a stream the application has closed while the peer had+--   not said where it ends.+data Departed = Departed+    { depHighest :: Int+    -- ^ 'rxHighest' so far+    , depLimit :: Int+    -- ^ The stream's receive limit, which nothing raises any more+    }+    deriving (Eq, Show)+ instance Show Stream where     show s = show $ streamId s @@ -80,19 +89,22 @@     streamStateTx   <- newIORef emptyStreamState     streamStateRx   <- newIORef emptyStreamState     streamRecvQ     <- newRecvStreamQ-    streamReass     <- newIORef (0, Skew.empty)-    streamSyncFinTx <- newEmptyMVar+    streamReass     <- newIORef $ Reass Map.empty Nothing+    streamSyncFinTx <- newEmptyTMVarIO+    streamResetTx   <- newIORef False     streamResetRx   <- newIORef Nothing     streamReleased  <- newIORef False-    streamRxBounds  <- newIORef emptyRxBounds+    streamRxBounds  <- newTVarIO emptyRxBounds     return Stream{..} {- FOURMOLU_ENABLE -}  syncFinTx :: Stream -> IO ()-syncFinTx s = void $ tryPutMVar (streamSyncFinTx s) ()+syncFinTx s = void $ atomically $ tryPutTMVar (streamSyncFinTx s) ()  waitFinTx :: Stream -> IO ()-waitFinTx s = takeMVar $ streamSyncFinTx s+waitFinTx s =+    atomically $+        takeTMVar (streamSyncFinTx s) `orElse` peerClosedSTM (streamConnection s)  ---------------------------------------------------------------- @@ -108,19 +120,13 @@     }     deriving (Eq, Show) -instance Frag RxStreamData where-    currOff r = rxstrmOff r-    nextOff r = rxstrmOff r + rxstrmLen r-    shrink off' (RxStreamData bs off len fin) =-        let n = off' - off-            bs' = BS.drop n bs-            len' = len - n-         in RxStreamData-                { rxstrmData = bs'-                , rxstrmOff = off'-                , rxstrmLen = len'-                , rxstrmFin = fin-                }+-- | What a stream holds past the point it has delivered in order.+data Reass = Reass+    { reassPieces :: Map Offset StreamData+    -- ^ Disjoint, and each starting past the delivered offset+    , reassFin :: Maybe Offset+    -- ^ Where the stream ends, once a FIN is held but not yet delivered+    }  ---------------------------------------------------------------- 

@@ -6,6 +6,7 @@     module Network.QUIC.Types.Ack,     module Network.QUIC.Types.CID,     module Network.QUIC.Types.Constants,+    module Network.QUIC.Types.Delay,     module Network.QUIC.Types.Error,     module Network.QUIC.Types.Exception,     module Network.QUIC.Types.Frame,@@ -20,6 +21,7 @@ import Network.QUIC.Types.Ack import Network.QUIC.Types.CID import Network.QUIC.Types.Constants+import Network.QUIC.Types.Delay import Network.QUIC.Types.Error import Network.QUIC.Types.Exception import Network.QUIC.Types.Frame

@@ -83,6 +83,24 @@         z = s - fromIntegral g - 2         lo = z - fromIntegral r +-- | Keeping only the newest ranges so that the ACK frame, with its type+--   and a one-octet ACK Delay, takes at most the given octets.  The+--   largest acknowledged always stays (RFC 9000 Sec 13.2.3).+capAckInfo :: Int -> AckInfo -> AckInfo+capAckInfo limit (AckInfo lpn fr grs) = AckInfo lpn fr $ take n grs+  where+    base = 2 + varIntLen lpn + varIntLen fr+    sizes = scanl1 (+) [varIntLen g + varIntLen r | (g, r) <- grs]+    fits k siz = base + varIntLen k + siz <= limit+    n = length $ takeWhile id $ zipWith fits [1 ..] sizes++varIntLen :: Int -> Int+varIntLen i+    | i < 0x40 = 1+    | i < 0x4000 = 2+    | i < 0x40000000 = 4+    | otherwise = 8+ fromAckInfoToPred :: AckInfo -> (PacketNumber -> Bool) fromAckInfoToPred (AckInfo lpn fr grs) =     \x -> any (f x) $ loop grs stt [(stt, lpn)]

@@ -0,0 +1,35 @@+-- | The ACK Delay field.  The constructor is not exported: the field is+--   only read and written here, and only turned into time with the+--   exponent of the ACK's sender.+module Network.QUIC.Types.Delay (+    Delay,+    noDelay,+    ackDelayMicro,+    encodeDelay,+    decodeDelay,+) where++import Network.QUIC.Imports+import Network.QUIC.Types.Integer+import Network.QUIC.Types.Time++-- | The ACK Delay field as sent, in units of 2 ^ ack_delay_exponent+--   microseconds of the ACK's sender (RFC 9000 Sec 19.3).+newtype Delay = Delay Int deriving (Eq, Show)++-- | An ACK Delay of zero, whatever the exponent.+noDelay :: Delay+noDelay = Delay 0++-- | Microseconds from the ACK Delay field and the sender's exponent,+--   saturating rather than overflowing on a value out of range.+ackDelayMicro :: Int -> Delay -> Microseconds+ackDelayMicro e (Delay d)+    | d > maxBound `shiftR` e = Microseconds maxBound+    | otherwise = Microseconds $ d `shiftL` e++encodeDelay :: WriteBuffer -> Delay -> IO ()+encodeDelay wbuf (Delay d) = encodeInt' wbuf $ fromIntegral d++decodeDelay :: ReadBuffer -> IO Delay+decodeDelay rbuf = Delay . fromIntegral <$> decodeInt' rbuf

@@ -32,6 +32,7 @@     | ExitConnection     | WrongTransportParameter     | WrongDatagramSizeParameter+    | Wrong0RTTLimitParameter     | WrongVersionInformation     | BreakForever     deriving (Eq, Show)

@@ -6,8 +6,8 @@ import Network.QUIC.Imports import Network.QUIC.Types.Ack import Network.QUIC.Types.CID+import Network.QUIC.Types.Delay import Network.QUIC.Types.Error-import Network.QUIC.Types.Time  ---------------------------------------------------------------- @@ -77,8 +77,6 @@ isUnidirectional :: StreamId -> Bool isUnidirectional sid = (0b10 .&. sid) == 2 -type Delay = Milliseconds- type Fin = Bool  type CryptoData = ByteString@@ -107,12 +105,4 @@ inFlight ConnectionCloseApp{} = False inFlight _                    = True -rateControled :: Frame -> Bool-rateControled ResetStream{}        = True-rateControled StopSending{}        = True-rateControled PathChallenge{}      = True-rateControled PathResponse{}       = True-rateControled NewConnectionID{}    = True-rateControled RetireConnectionID{} = True-rateControled _                    = False {- FOURMOLU_ENABLE -}

@@ -7,6 +7,7 @@ import Codec.Serialise import Data.Ix import GHC.Generics+import Network.Socket (SockAddr) import Network.TLS.QUIC (ExtensionID (EID_QuicTransportParameters, ExtensionID)) import Text.Printf @@ -182,6 +183,8 @@     , rpTimeRecevied :: TimeMicrosecond     , rpReceivedBytes :: Int     , rpEncryptionLevel :: EncryptionLevel+    , rpPeerAddr :: Maybe SockAddr+    -- ^ Where a server received it from     }     deriving (Eq, Show) @@ -193,6 +196,7 @@         , rpTimeRecevied = tim         , rpReceivedBytes = bytes         , rpEncryptionLevel = lvl+        , rpPeerAddr = Nothing         }  ----------------------------------------------------------------

@@ -1,4 +1,6 @@+/* watch_socket: 0 for an event, -1 to try again (EINTR), -2 to give up. */ #if defined(OS_MacOS)+#include <errno.h> #include <sys/socket.h> #include <sys/types.h> #include <unistd.h>@@ -9,11 +11,15 @@    struct kev_request filter = { 0 };     int s = socket(PF_SYSTEM, SOCK_RAW, SYSPROTO_EVENT);+   if (s < 0) return -1;     filter.vendor_code = KEV_VENDOR_APPLE;    filter.kev_class = KEV_NETWORK_CLASS;    filter.kev_subclass = KEV_DL_SUBCLASS;-   ioctl(s, SIOCSKEVFILT, &filter);+   if (ioctl(s, SIOCSKEVFILT, &filter) < 0) {+     close(s);+     return -1;+   }     return s; }@@ -22,7 +28,7 @@    struct kern_event_msg msg;    int ret = recv(s, &msg, sizeof(msg), 0);    if (ret < 0) {-     return -1;+     return errno == EINTR ? -1 : -2;    } else {      return 0;    }@@ -33,6 +39,7 @@   return ret; } #elif defined(OS_Linux)+#include <errno.h> #include <sys/socket.h> #include <sys/types.h> #include <unistd.h>@@ -43,10 +50,14 @@   struct sockaddr_nl filter = { 0 };    int s = socket(AF_NETLINK, SOCK_DGRAM, NETLINK_ROUTE);+  if (s < 0) return -1;    filter.nl_family = AF_NETLINK;   filter.nl_groups = RTMGRP_IPV4_ROUTE | RTMGRP_IPV6_ROUTE;-  bind(s, (struct sockaddr *)&filter, sizeof(filter));+  if (bind(s, (struct sockaddr *)&filter, sizeof(filter)) < 0) {+    close(s);+    return -1;+  }    return s; }@@ -55,7 +66,7 @@   char msg[4096] = {0};   int ret = recv(s, msg, sizeof(msg), 0);    if (ret < 0) {-     return -1;+     return errno == EINTR ? -1 : -2;    } else {      return 0;    }@@ -66,7 +77,7 @@   return ret; } #else-int open_socket () { return 0; }+int open_socket () { return -1; }  int watch_socket(int s) { return -2; } 

@@ -1,6 +1,6 @@ cabal-version:      2.0 name:               quic-version:            0.3.19+version:            0.3.20 license:            BSD3 license-file:       LICENSE maintainer:         kazu@iij.ad.jp@@ -39,6 +39,7 @@     other-modules:         Network.QUIC.Client.Reader         Network.QUIC.Client.Run+        Network.QUIC.Client.WatchDog         Network.QUIC.Closer         Network.QUIC.Common         Network.QUIC.Config@@ -48,6 +49,7 @@         Network.QUIC.Connection.Misc         Network.QUIC.Connection.PacketNumber         Network.QUIC.Connection.Queue+        Network.QUIC.Connection.RateControl         Network.QUIC.Connection.Role         Network.QUIC.Connection.State         Network.QUIC.Connection.Stream@@ -97,11 +99,9 @@         Network.QUIC.Server.Run         Network.QUIC.Socket         Network.QUIC.Stream-        Network.QUIC.Stream.Frag         Network.QUIC.Stream.Misc         Network.QUIC.Stream.Queue         Network.QUIC.Stream.Reass-        Network.QUIC.Stream.Skew         Network.QUIC.Stream.Table         Network.QUIC.Stream.Types         Network.QUIC.TLS@@ -109,6 +109,7 @@         Network.QUIC.Types.Ack         Network.QUIC.Types.CID         Network.QUIC.Types.Constants+        Network.QUIC.Types.Delay         Network.QUIC.Types.Error         Network.QUIC.Types.Exception         Network.QUIC.Types.Frame@@ -139,7 +140,7 @@         fast-logger >=3.2.8 && <3.3,         filepath,         iproute >=1.7.12 && <1.8,-        network >=3.2.10,+        network >=3.2.12,         network-byte-order >=0.1.7 && <0.2,         network-control >=0.1.5 && <0.2,         ram,@@ -147,7 +148,7 @@         serialise,         stm >=2.5 && <2.6,         time-manager >=0.5 && <0.6,-        tls >=2.4.0 && <2.5,+        tls >=2.4.0 && <2.6,         unix-time >=0.4.12 && <0.6      if os(linux)@@ -246,6 +247,7 @@         TokenSpec         TransportError         TypesSpec+        WatchDogSpec      default-language:   Haskell2010     default-extensions: Strict StrictData@@ -267,7 +269,7 @@         fast-logger >=3.2.8 && <3.3,         filepath,         hspec,-        network >=3.2.10,+        network >=3.2.12,         quic,         ram,         stm,

@@ -27,6 +27,7 @@ import Data.IORef import qualified Data.List as L import qualified Data.List.NonEmpty as NE+import Data.Maybe (isJust, isNothing) import Network.Socket import Network.Socket.ByteString import Network.TLS hiding (Version)@@ -159,6 +160,16 @@       DelayClientPacket Int     | -- | Hold the n-th datagram from the server back for a second.       DelayServerPacket Int+    | -- | Once the MVar is filled, send the client, as if from the server,+      --   what the function makes of the client's and the server's CIDs.+      InjectToClient (MVar ()) (CID -> CID -> IO ByteString)+    | -- | Record the size of every short-header datagram, with whether it+      --   came from the client.+      RecordSizes (IORef [(Bool, Int)])+    | -- | Once the MVar is filled, send the client's next short-header+      --   datagram to the server from another socket, which answers+      --   nothing: an on-path attacker's spurious migration.+      ForwardAside (MVar ())  withPipe :: Scenario -> IO () -> IO () withPipe = withPipeWith False@@ -196,44 +207,48 @@     irefC <- newIORef 0     irefS <- newIORef 0     E.bracket (openSocket addrC) close $ \sockC ->-        E.bracket (openSocket addrS) close $ \sockS -> do-            setSocketOption sockC ReuseAddr 1-            setSocketOption sockS ReuseAddr 1-            bind sockC saC-            -- Bound, not connected.  A connected UDP socket turns an ICMP-            -- port-unreachable from the peer into ECONNREFUSED on the next-            -- operation, and the peers here come and go with every test, so-            -- the relay was being killed by a reply to something it had sent-            -- to an address that had just closed.  It surfaced on the-            -- receive, not the send:-            ---            --   DIED C->S: Network.Socket.recvBuf: does not exist-            --              (Connection refused)-            ---            -- and since these are forkIO threads with nobody watching, the-            -- relay simply stopped.  The client then sent Initial packets-            -- until the idle timeout and heard nothing.-            addrAny <- resolve "0"-            bind sockS $ addrAddress addrAny-            when stray $-                E.bracket (openSocket addrC) close $ \sock -> do-                    void $ sendTo sock (BS.pack [0x40, 1, 2, 3]) saC-                    void $ sendTo sock (BS.pack [0xc0, 1, 2, 3]) saC+        E.bracket (openSocket addrS) close $ \sockS ->+            E.bracket (openSocket addrS) close $ \sockA -> do+                setSocketOption sockC ReuseAddr 1+                setSocketOption sockS ReuseAddr 1+                bind sockC saC+                -- Bound, not connected.  A connected UDP socket turns an ICMP+                -- port-unreachable from the peer into ECONNREFUSED on the next+                -- operation, and the peers here come and go with every test, so+                -- the relay was being killed by a reply to something it had sent+                -- to an address that had just closed.  It surfaced on the+                -- receive, not the send:+                --+                --   DIED C->S: Network.Socket.recvBuf: does not exist+                --              (Connection refused)+                --+                -- and since these are forkIO threads with nobody watching, the+                -- relay simply stopped.  The client then sent Initial packets+                -- until the idle timeout and heard nothing.+                addrAny <- resolve "0"+                bind sockS $ addrAddress addrAny+                bind sockA $ addrAddress addrAny+                when stray $+                    E.bracket (openSocket addrC) close $ \sock -> do+                        void $ sendTo sock (BS.pack [0x40, 1, 2, 3]) saC+                        void $ sendTo sock (BS.pack [0xc0, 1, 2, 3]) saC -            -- The relaying threads have to stop before the sockets close.-            -- Run at the end of body instead, the kills are skipped whenever-            -- body throws, and the threads are then left in recv on a socket-            -- the bracket has just closed.  That surfaces as "threadWait:-            -- invalid argument (Bad file descriptor)" from a thread nobody is-            -- watching, and buries whatever the test was really failing on.-            E.bracket (startRelay saS sockC sockS irefC irefS) stopRelay $ \_ -> body+                -- The relaying threads have to stop before the sockets close.+                -- Run at the end of body instead, the kills are skipped whenever+                -- body throws, and the threads are then left in recv on a socket+                -- the bracket has just closed.  That surfaces as "threadWait:+                -- invalid argument (Bad file descriptor)" from a thread nobody is+                -- watching, and buries whatever the test was really failing on.+                E.bracket (startRelay saS sockC sockS sockA irefC irefS) stopRelay $ \_ -> body   where-    startRelay saS sockC sockS irefC irefS = do+    startRelay saS sockC sockS sockA irefC irefS = do         stop <- newTVarIO False         -- Where to send what comes back from the server.  Set once the client         -- has introduced itself, which is before the server can have anything         -- to say about it.         peerRef <- newIORef Nothing+        -- The CIDs of the server's first long-header packet.+        cidsRef <- newIORef Nothing         -- from client         tid0 <- forkIO $ do             -- Wait for the client to introduce itself, and take the first@@ -265,42 +280,72 @@             -- leaves a client behind breaks the next one however this             -- chooses, so a test must not leave one behind.             mhello <- waitForClientHello stop sockC-            forM_ mhello $ \(bs, saO) -> do-              writeIORef peerRef $ Just saO-              n0 <- atomicModifyIORef' irefC $ \x -> (x + 1, x)-              dropPacket0 <- shouldDrop scenario True n0-              unless dropPacket0 $ void $ sendTo sockS bs saS-              relayLoop stop sockC $ \(bs1, sa) -> do-                -- Only from the client we latched onto.  The connect this-                -- replaces did that in the kernel; doing it here keeps-                -- leftovers from the connection that just closed from being-                -- counted, which would shift the drop indices.-                when (sa == saO) $ do-                    n <- atomicModifyIORef' irefC $ \x -> (x + 1, x)-                    dropPacket <- shouldDrop scenario True n-                    let isCC = BS.length bs1 < 200-                    when (isCC || not dropPacket) $-                        delayIf (shouldDelay scenario True n) $-                            void $-                                sendTo sockS bs1 saS+            case mhello of+                Nothing -> return ()+                Just (bs, saO) -> do+                    writeIORef peerRef $ Just saO+                    n0 <- atomicModifyIORef' irefC $ \x -> (x + 1, x)+                    dropPacket0 <- shouldDrop scenario True n0+                    unless dropPacket0 $ void $ sendTo sockS bs saS+                    relayLoop stop sockC $ \(bs1, sa) -> do+                        -- Only from the client we latched onto.  The connect this+                        -- replaces did that in the kernel; doing it here keeps+                        -- leftovers from the connection that just closed from being+                        -- counted, which would shift the drop indices.+                        when (sa == saO) $ do+                            recordSize scenario True bs1+                            n <- atomicModifyIORef' irefC $ \x -> (x + 1, x)+                            dropPacket <- shouldDrop scenario True n+                            let isCC = BS.length bs1 < 200+                            aside <- case scenario of+                                ForwardAside trigger+                                    | BS.head bs1 .&. 0x80 == 0 ->+                                        isJust <$> tryTakeMVar trigger+                                _ -> return False+                            let sock = if aside then sockA else sockS+                            when (isCC || not dropPacket) $+                                delayIf (shouldDelay scenario True n) $+                                    void $+                                        sendTo sock bs1 saS         -- from server         tid1 <- forkIO $ relayLoop stop sockS $ \(bs, _) -> do+            mcids <- readIORef cidsRef+            when (isNothing mcids) $ writeIORef cidsRef $ longHeaderCIDs bs+            recordSize scenario False bs             n <- atomicModifyIORef' irefS $ \x -> (x + 1, x)             dropPacket <- shouldDrop scenario False n             let isCC = BS.length bs < 200             when (isCC || not dropPacket) $ do                 mpeer <- readIORef peerRef-                forM_ mpeer $ \sa ->-                    delayIf (shouldDelay scenario False n) $ void $ sendTo sockC bs sa-        return (stop, tid0, tid1)+                case mpeer of+                    Nothing -> return ()+                    Just sa ->+                        delayIf (shouldDelay scenario False n) $ void $ sendTo sockC bs sa+        tid2 <- forkIO $ case scenario of+            InjectToClient trigger mk -> do+                takeMVar trigger+                Just (dcid, scid) <- readIORef cidsRef+                Just sa <- readIORef peerRef+                bs <- mk dcid scid+                void $ sendTo sockC bs sa+            _ -> return ()+        return (stop, tid0, tid1, tid2)     -- Telling these threads to stop rather than killing them: a thread     -- blocked in a socket call on Windows is not reachable by 'killThread',     -- and the brackets close the sockets straight after.  The kill stays     -- for a thread which is somewhere else.-    stopRelay (stop, tid0, tid1) = do+    stopRelay (stop, tid0, tid1, tid2) = do         atomically $ writeTVar stop True         killThread tid0         killThread tid1+        killThread tid2+    longHeaderCIDs bs+        | BS.length bs < 7 || BS.head bs .&. 0x80 == 0 = Nothing+        | otherwise =+            let dlen = fromIntegral $ BS.index bs 5+                (dcid, rest) = BS.splitAt dlen $ BS.drop 6 bs+                slen = fromIntegral $ BS.head rest+             in Just (toCID dcid, toCID $ BS.take slen $ BS.drop 1 rest)     relayRecv stop sock = recvFromWith sock 2048 (check =<< readTVar stop)     relayLoop stop sock act = loop       where@@ -336,6 +381,10 @@         | fromC = return False         | otherwise = return (pn `elem` ns)     shouldDrop _ _ _ = return False+    recordSize (RecordSizes ref) fromC bs+        | not (BS.null bs) && BS.head bs .&. 0x80 == 0 =+            atomicModifyIORef' ref $ \xs -> ((fromC, BS.length bs) : xs, ())+    recordSize _ _ _ = return ()     shouldDelay (DelayClientPacket k) fromC pn = fromC && pn == k     shouldDelay (DelayServerPacket k) fromC pn = not fromC && pn == k     shouldDelay _ _ _ = False

@@ -7,6 +7,9 @@ import qualified Control.Exception as E import Control.Monad import qualified Data.ByteString as BS+import Data.IORef+import qualified Network.Socket as NS+import qualified Network.Socket.ByteString as NSB import Network.TLS (Group (..), HandshakeMode13 (..)) import qualified Network.TLS as TLS import qualified System.Timeout as Timeout@@ -95,6 +98,12 @@                                 }                         }             test0RTTFlowControl cc sc waitS+        it "keeps a resumed connection to the limits the server gives now" $ do+            testLoweredLimits testClientConfig sc0 waitS+                `shouldReturn` (PreSharedKey, 3)+        it "refuses 0-RTT accepted with the limits lowered" $ do+            testLowered0RTT testClientConfig sc0{scUse0RTT = True} waitS+                `shouldThrow` protocolViolation         it "sends 0-RTT data without waiting for the handshake" $ do             let sc = sc0{scUse0RTT = True}             test0RTTSendsEarly sc waitS@@ -119,6 +128,26 @@                         te == cryptoError TLS.HandshakeFailure                     | otherwise = False             testHandshake3 cc1 cc2 sc waitS handshakeFailure+        it "fails with no version in common" $ do+            let cc1 = testClientConfig{ccVersion = Version1, ccVersions = [Version1]}+                cc2 = testClientConfig+                sc = sc0{scVersions = [Version2]}+                failed VersionNegotiationFailed = True+                failed _ = False+            testHandshake3 cc1 cc2 sc waitS failed+        it "keeps serving after datagrams too large to receive" $ do+            mvar <- newEmptyMVar+            let serv conn = waitEstablished conn >> putMVar mvar ()+            E.bracket (forkIO $ S.run sc0 serv) killThread $ \_ -> do+                waitS+                -- Winsock fails a receive into a buffer too small for the+                -- datagram, where POSIX truncates it.+                sendOversized 3+                r <- Timeout.timeout 5000000 $+                    C.run testClientConfig $ \conn -> do+                        waitEstablished conn+                        takeMVar mvar+                r `shouldBe` Just ()         it "can handshake with large HE from a client" $ do             let cc0 = testClientConfig                 params =@@ -137,6 +166,14 @@                 sc = sc0{scParameters = params}             testHandshake cc sc waitS FullHandshake +-- | Datagrams larger than the server's receive buffer, to the test server.+sendOversized :: Int -> IO ()+sendOversized n = do+    let hints = NS.defaultHints{NS.addrSocketType = NS.Datagram}+    addr : _ <- NS.getAddrInfo (Just hints) (Just "127.0.0.1") (Just "15003")+    E.bracket (NS.openSocket addr) NS.close $ \s ->+        replicateM_ n $ NSB.sendTo s (BS.replicate 4000 0) (NS.addrAddress addr)+ onE :: IO b -> IO a -> IO a onE h b = E.onException b h @@ -327,3 +364,72 @@         sendStream s "bye"         closeStream s         putMVar mvar ()++-- | A first connection to a server allowing a hundred streams, for its+--   resumption information.+resumeFrom :: ClientConfig -> ServerConfig -> IO () -> IO ResumptionInfo+resumeFrom cc sc waitS =+    E.bracket (forkIO $ S.run sc1 serv) killThread $ \_ -> do+        waitS+        C.run cc $ \conn -> do+            s <- stream conn+            sendStream s "first"+            shutdownStream s+            _ <- recvStream s 1024+            threadDelay 50000+            getResumptionInfo conn+  where+    sc1 = sc{scParameters = (scParameters sc){initialMaxStreamsBidi = 100}}+    serv conn = do+        s <- acceptStream conn+        _ <- recvStream s 1024+        sendStream s "bye"+        closeStream s++-- | The same server, as far as its session tickets go, then allowing+--   three streams.  RFC 9000 Sec 7.4.1 has the client keep to what it+--   remembers only for 0-RTT.  The limits only ever rose on the client, so+--   a resumed connection went on opening streams up to the hundred, and the+--   server closed it with STREAM_LIMIT_ERROR.+--+-- Answers how the second connection went and how many streams it opened+-- in a second and a half.+testLoweredLimits+    :: ClientConfig -> ServerConfig -> IO () -> IO (HandshakeMode13, Int)+testLoweredLimits cc1 sc waitS = do+    res <- resumeFrom cc1 sc waitS+    threadDelay 100000+    E.bracket (forkIO $ S.run sc2 serv) killThread $ \_ -> do+        waitS+        C.run cc1{ccResumption = res} $ \conn -> do+            opened <- newIORef (0 :: Int)+            _ <- Timeout.timeout 1500000 $ forever $ do+                s <- stream conn+                sendStream s "x"+                atomicModifyIORef' opened $ \x -> (x + 1, ())+            (,) <$> (handshakeMode <$> getConnectionInfo conn) <*> readIORef opened+  where+    sc2 = sc{scParameters = (scParameters sc){initialMaxStreamsBidi = 3}}+    serv conn = forever $ void $ acceptStream conn++-- | As above, but with 0-RTT, which the server accepts with fifty streams+--   where it gave a hundred before: RFC 9000 Sec 7.4.1 says it MUST NOT.+--   The client opens one stream, within either, so that it is the client+--   that notices.+testLowered0RTT :: ClientConfig -> ServerConfig -> IO () -> IO ()+testLowered0RTT cc1 sc waitS = do+    res <- resumeFrom cc1 sc waitS+    threadDelay 100000+    E.bracket (forkIO $ S.run sc2 serv) killThread $ \_ -> do+        waitS+        C.run cc1{ccResumption = res, ccUse0RTT = True} $ \conn -> do+            s <- stream conn+            sendStream s "x"+            threadDelay 1000000+  where+    sc2 = sc{scParameters = (scParameters sc){initialMaxStreamsBidi = 50}}+    serv conn = forever $ void $ acceptStream conn++protocolViolation :: QUICException -> Bool+protocolViolation (TransportErrorIsSent ProtocolViolation _) = True+protocolViolation _ = False

@@ -6,10 +6,17 @@  import Control.Concurrent import Control.Concurrent.Async+import Control.Concurrent.STM (readTVarIO) import qualified Control.Exception as E import Control.Monad+import Data.Bits (shiftR) import qualified Data.ByteString as BS+import qualified Data.ByteString.Short as Short import Data.IORef+import qualified Data.IntMap.Strict as IntMap+import GHC.Conc (getUncaughtExceptionHandler, setUncaughtExceptionHandler)+import qualified Network.Socket as NS+import qualified Network.Socket.ByteString as NSB import qualified System.Timeout as Timeout import Test.Hspec @@ -40,6 +47,34 @@             Timeout.timeout 5000000 (takeMVar var) >>= \r -> case r of                 Just () -> return ()                 Nothing -> expectationFailure "server never became ready"+    -- RFC 9000 Sec 6.2 and 17.2.5.2: once a packet from the server has been+    -- processed, a Version Negotiation or Retry packet is to be discarded.+    -- The connection has moved to Version2 by compatible negotiation.+    describe "a forged packet after the handshake" $ do+        it "ignores Version Negotiation" $+            testInjected cc sc waitS $ \dcid scid ->+                encodeVersionNegotiationPacket $+                    VersionNegotiationPacket dcid scid [Version1]+        it "ignores Retry" $+            testInjected cc sc waitS $ \dcid scid ->+                encodeRetryPacket $+                    RetryPacket Version2 dcid (toCID "forged") "token" (Left scid)+    describe "max_udp_payload_size" $ do+        it "bounds the datagrams each side sends" $ do+            let small p = p{maxUdpPayloadSize = 1200}+                cc' = cc{ccParameters = small $ ccParameters cc}+                sc' = sc{scParameters = small $ scParameters sc}+            ref <- newIORef []+            withPipe (RecordSizes ref) $ testEcho cc' sc' waitS 20+            sizes <- readIORef ref+            [siz | (True, siz) <- sizes] `shouldSatisfy` all (<= 1200)+            [siz | (False, siz) <- sizes] `shouldSatisfy` all (<= 1200)+            -- Data went both ways in full-sized packets.+            [siz | (True, siz) <- sizes] `shouldSatisfy` elem 1200+            [siz | (False, siz) <- sizes] `shouldSatisfy` elem 1200+    describe "STOP_SENDING in a burst" $ do+        it "resets every stream (20)" $ testStopMany cc sc waitS 20+        it "resets every stream (40)" $ testStopMany cc sc waitS 40     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@@ -106,6 +141,8 @@         -- 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 "sends RESET_STREAM with the bytes queued as final size" $ do+            withPipe (DropClientPacket []) $ testResetAfterSend 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@@ -118,6 +155,67 @@             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 "a stream closed before the peer finishes it" $ do+        it "is forgotten once the peer answers STOP_SENDING" $ do+            withPipe (DropClientPacket []) $ testDepartedForgotten cc sc waitS+        it "is kept within the limit when the peer does not answer" $ do+            withPipe (DropClientPacket []) $ testDepartedBounded cc sc waitS+        it "is forgotten when its FIN crosses the close" $ do+            withPipe (DropClientPacket []) $ testFinCrossesClose cc sc waitS+    describe "the server's path" $ do+        it "stays put for a short header packet that does not decrypt" $+            testForgedPath cc sc waitS $ \_ (CID dcid) ->+                BS.concat [BS.singleton 0x40, Short.fromShort dcid, BS.replicate 1100 0x11]+        it "stays put for a Handshake packet that does not decrypt" $+            testForgedPath cc sc waitS $ \ver (CID dcid) ->+                let Version v = ver+                    typ = if ver == Version2 then 0x30 else 0x20+                 in BS.concat+                        [ BS.singleton (0xc0 + typ)+                        , BS.pack [fromIntegral (v `shiftR` n) | n <- [24, 16, 8, 0]]+                        , BS.singleton (fromIntegral $ Short.length dcid)+                        , Short.fromShort dcid+                        , BS.pack [0, 0x44, 0x00]+                        , BS.replicate 1024 0x11+                        ]+        it "follows a client that rebinds" $ testRebindPath cc sc waitS+        it "goes back when the new one is not validated" $+            testSpuriousMigration cc sc waitS+    describe "a stateless reset" $ do+        it "is smaller than the packet it answers" $ do+            withAsync (run sc $ \_ -> return ()) $ \_ -> do+                waitS+                -- A reset of the server's, sent back to it, as another+                -- server would.  It used to be answered at the same size.+                sizes <- resetSizes [1280, 640, 60]+                sizes `shouldBe` [Just 640, Just 320, Just 30]+    describe "the client's socket" $ do+        it "is closed when the peer ends the connection" $ do+            withPipe (DropClientPacket []) $ testSocketClosed cc sc waitS+    describe "the peer closing with NO_ERROR" $ do+        it "wakes a client waiting in recvStream" $ do+            withPipe (DropClientPacket []) $+                testPeerClosed cc sc waitS Nothing $ \conn -> do+                    strm <- stream conn+                    sendStream strm "ping"+                    void $ recvStream strm 1024+                    recvStream strm 1024+        it "wakes a client waiting in acceptStream" $ do+            withPipe (DropClientPacket []) $+                testPeerClosed cc sc waitS Nothing $ \conn -> do+                    strm <- stream conn+                    sendStream strm "ping"+                    void $ recvStream strm 1024+                    void $ acceptStream conn+                    return ""+        it "lets a client that is not waiting finish" $ do+            withPipe (DropClientPacket []) $+                testPeerClosed cc sc waitS (Just "done") $ \conn -> do+                    strm <- stream conn+                    sendStream strm "ping"+                    void $ recvStream strm 1024+                    threadDelay 300000+                    return "done"     describe "stream state" $ do         it "can still send after the peer's RESET_STREAM" $ do             withPipe (DropClientPacket []) $ testSendAfterReset cc sc waitS@@ -147,11 +245,23 @@             withPipe (DropClientPacket []) $ testCloseWhileArriving cc sc waitS         it "counts a reset stream's final size against the connection" $ do             withPipe (DropClientPacket []) $ testResetCountsAgainstTheWindow cc sc waitS+        it "holds a reset stream's final size to the stream's limit" $ do+            withPipe (DropClientPacket []) $ testResetOverStreamLimit cc sc waitS+        it "counts the gaps of a stream against the connection" $ do+            withPipe (DropClientPacket []) $ testGapCountsAgainstTheWindow cc sc waitS+        it "holds a closed stream to the stream's limit" $ do+            withPipe (DropClientPacket []) $+                testDepartedOverLimit (100000, 1000) cc sc waitS+        it "holds a closed stream to the connection's limit" $ do+            withPipe (DropClientPacket []) $+                testDepartedOverLimit (1000, 100000) 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+        it "leaves nothing uncaught when stopped as a connection ends" $ do+            withPipe (DropClientPacket []) $ noUncaught $ testSlowApp cc sc waitS     describe "port handover" $ do         it "ignores a leftover datagram from the connection that just closed" $             withPipeStray (Randomly 20) $@@ -242,6 +352,46 @@         sendStream strm payload         takeMVar doneVar +-- | A reset straight after 'sendStream' finds most of the data still+--   queued.  The final size has to cover it, and none of it may follow the+--   RESET_STREAM: either makes the peer close with FINAL_SIZE_ERROR.+testResetAfterSend :: C.ClientConfig -> ServerConfig -> IO () -> IO ()+testResetAfterSend cc0 sc waitS = do+    finalSizeVar <- newEmptyMVar+    doneVar <- newEmptyMVar+    let hooks = (ccHooks cc0){onResetStreamReceived2 = record finalSizeVar}+        cc = cc0{ccHooks = hooks}+    withAsync (server doneVar) $ \_ -> client cc finalSizeVar doneVar+  where+    payload = BS.replicate 100000 0+    record finalSizeVar _strm _aerr finalSize = void $ tryPutMVar finalSizeVar finalSize++    client cc finalSizeVar doneVar = do+        waitS+        C.run cc $ \conn -> do+            strm <- stream conn+            sendStream strm "open"+            mres <- Timeout.timeout 5000000 $ takeMVar finalSizeVar+            mres `shouldBe` Just (BS.length payload)+            -- The connection is still there once anything sent after the+            -- RESET_STREAM would have arrived.+            strm2 <- stream conn+            sendStream strm2 "ping"+            shutdownStream strm2+            recvStream strm2 4 `shouldReturn` "ping"+            putMVar doneVar ()++    server doneVar = run sc $ \conn -> do+        strm <- acceptStream conn+        consumeBytes strm 4+        sendStream strm payload+        resetStream strm $ ApplicationProtocolError 0+        strm2 <- acceptStream conn+        consumeBytes strm2 4+        sendStream strm2 "ping"+        shutdownStream strm2+        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@@ -348,6 +498,105 @@                     o <- atomicModifyIORef' openRef $ \x -> (x + 1, x + 1)                     atomicModifyIORef' maxOpen $ \m -> (max m o, ()) +-- | The server reads a byte of each request and closes the stream, the+--   client never sends a FIN.  What the server kept of each, to give back+--   the window late frames spend, stayed for as long as the connection did;+--   STOP_SENDING now has the client reset them, which settles them.+testDepartedForgotten :: C.ClientConfig -> ServerConfig -> IO () -> IO ()+testDepartedForgotten cc sc waitS = do+    left <- newEmptyMVar+    withAsync (server left) $ \_ -> do+        waitS+        C.run cc $ \conn -> do+            replicateM_ n $ do+                strm <- stream conn+                sendStream strm "x"+            takeMVar left `shouldReturn` 0+  where+    n = 300 :: Int+    server left = run sc $ \conn -> do+        replicateM_ n $ do+            strm <- acceptStream conn+            _ <- recvStream strm 1+            closeStream strm+        let departed = IntMap.size <$> readTVarIO (departedStreams conn)+            wait :: Int -> IO Int+            wait 0 = departed+            wait k = do+                d <- departed+                if d == 0 then return 0 else threadDelay 10000 >> wait (k - 1)+        wait 300 >>= putMVar left++-- | As above, but the client does not hear the STOP_SENDING.  The streams+--   it leaves unfinished are not given back to its limit, so the server+--   keeps no more of them than it allows open at once.+testDepartedBounded :: C.ClientConfig -> ServerConfig -> IO () -> IO ()+testDepartedBounded cc sc0 waitS = do+    most <- newIORef (0 :: Int)+    opened <- newIORef (0 :: Int)+    withAsync (server most) $ \_ -> do+        waitS+        C.run cc $ \conn -> do+            _ <- Timeout.timeout 1500000 $ forever $ do+                strm <- stream conn+                sendStream strm "x"+                atomicModifyIORef' opened $ \x -> (x + 1, ())+            return ()+        readIORef opened >>= (`shouldSatisfy` (<= limit))+        readIORef most >>= (`shouldSatisfy` (<= limit))+  where+    limit = initialMaxStreamsBidi $ scParameters sc0+    sc = sc0{scHooks = (scHooks sc0){onPlainCreated = hide}}+    hide _ plain = plain{plainFrames = map unstop $ plainFrames plain}+    unstop StopSending{} = Ping+    unstop f = f+    server most = run sc $ \conn -> forever $ do+        strm <- acceptStream conn+        _ <- recvStream strm 1+        closeStream strm+        d <- IntMap.size <$> readTVarIO (departedStreams conn)+        atomicModifyIORef' most $ \m -> (max m d, ())++-- | The server closes each stream as soon as it has read a byte, while the+--   client's FIN for it is on its way.  The FIN is the only thing that ends+--   the stream here: the server's STOP_SENDING is hidden.  Every stream is+--   to be forgotten, and given back to the client's limit, in the end.+--+-- Closing used to take the stream out of the table and keep what it owed+-- in two steps, and a frame handled in between was counted in neither.  The+-- window was a few microseconds and this does not hit it; it checks that+-- FINs crossing the close, now one transaction against it, still settle.+testFinCrossesClose :: C.ClientConfig -> ServerConfig -> IO () -> IO ()+testFinCrossesClose cc sc0 waitS = do+    left <- newEmptyMVar+    withAsync (server left) $ \_ -> do+        waitS+        r <- Timeout.timeout 20000000 $ C.run cc $ \conn -> do+            replicateM_ n $ do+                strm <- stream conn+                sendStream strm "x"+                shutdownStream strm+            takeMVar left+        r `shouldBe` Just 0+  where+    n = 2000 :: Int+    sc = sc0{scHooks = (scHooks sc0){onPlainCreated = hide}}+    hide _ plain = plain{plainFrames = map unstop $ plainFrames plain}+    unstop StopSending{} = Ping+    unstop f = f+    server left = run sc $ \conn -> do+        replicateM_ n $ do+            strm <- acceptStream conn+            _ <- recvStream strm 1+            closeStream strm+        let departed = IntMap.size <$> readTVarIO (departedStreams conn)+            wait :: Int -> IO Int+            wait 0 = departed+            wait k = do+                d <- departed+                if d == 0 then return 0 else threadDelay 10000 >> wait (k - 1)+        wait 300 >>= putMVar left+ -- | 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.@@ -444,6 +693,99 @@         assertEndOfStream strm         putMVar mvar () +-- | The client sends @times@ blocks and the server sends them back.+testEcho :: C.ClientConfig -> ServerConfig -> IO () -> Int -> IO ()+testEcho cc sc waitS times = do+    mvar <- newEmptyMVar+    withAsync (server mvar) $ \_ -> client mvar+  where+    len = 10000 * times+    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+            consumeBytes strm len+            assertEndOfStream strm+            putMVar mvar ()+    -- Until the client has it all: returning closes the connection, and+    -- the client was then now and then reset before it had read it.+    server mvar = run sc $ \conn -> do+        strm <- acceptStream conn+        consumeBytes strm len+        assertEndOfStream strm+        replicateM_ times $ sendStream strm $ BS.replicate 10000 0+        shutdownStream strm+        takeMVar mvar++-- | Exchanging data after the client has been sent a forged packet.  The+--   client is to run its action once, to the end.+testInjected+    :: C.ClientConfig+    -> ServerConfig+    -> IO ()+    -> (CID -> CID -> IO BS.ByteString)+    -> IO ()+testInjected cc sc waitS mk = do+    trigger <- newEmptyMVar+    runs <- newIORef (0 :: Int)+    mvar <- newEmptyMVar+    r <- Timeout.timeout 10_000_000 $+        withPipe (InjectToClient trigger mk) $+            withAsync (server mvar) $+                \_ -> client trigger runs mvar+    r `shouldBe` Just ()+    readIORef runs `shouldReturn` 1+  where+    client trigger runs mvar = do+        waitS+        C.run cc $ \conn -> do+            atomicModifyIORef' runs $ \n -> (n + 1, ())+            waitEstablished conn+            void $ tryPutMVar trigger ()+            threadDelay 100_000+            strm <- stream conn+            sendStream strm $ BS.replicate 10000 0+            shutdownStream strm+            takeMVar mvar+    server mvar = run sc $ \conn -> do+        strm <- acceptStream conn+        consumeBytes strm 10000+        assertEndOfStream strm+        putMVar mvar ()++-- | The server asks the client to stop sending on @n@ streams at once.+testStopMany :: C.ClientConfig -> ServerConfig -> IO () -> Int -> IO ()+testStopMany cc sc waitS n = do+    mvar <- newEmptyMVar+    ref <- newIORef 0+    r <- Timeout.timeout 10_000_000 $+        withPipe (DropServerPacket []) $+            withAsync (server mvar) $+                \_ -> client mvar >>= writeIORef ref+    r `shouldBe` Just ()+    readIORef ref `shouldReturn` n+  where+    client mvar = do+        waitS+        C.run cc $ \conn -> do+            strms <- replicateM n $ do+                strm <- stream conn+                sendStream strm "x"+                return strm+            () <- takeMVar mvar+            threadDelay 300_000+            rs <- forM strms $ \strm ->+                E.try (sendStream strm "y") :: IO (Either QUICException ())+            return $ length [() | Left _ <- rs]+    server mvar = run sc $ \conn -> do+        strms <- replicateM n $ acceptStream conn+        mapM_ (\strm -> stopStream strm $ ApplicationProtocolError 0) strms+        putMVar mvar ()+        threadDelay 1_000_000+ testMultiSendRecv :: C.ClientConfig -> ServerConfig -> IO () -> Int -> IO () testMultiSendRecv cc sc waitS times = do     mvars <- replicateM concurrency newEmptyMVar@@ -734,6 +1076,100 @@         waitS         C.run cc $ \_conn -> threadDelay 1000000 +-- | RFC 9000, section 4.5: the final size counts against the stream's+--   limit as well as the connection's.  Only the connection's was checked,+--   so a reset could claim far more than the stream was ever allowed.+testResetOverStreamLimit :: C.ClientConfig -> ServerConfig -> IO () -> IO ()+testResetOverStreamLimit cc0 sc0 waitS =+    withAsync quietServer $ \_ -> client `shouldThrow` flowControlError+  where+    sc =+        sc0+            { scParameters =+                (scParameters sc0)+                    { initialMaxData = 1000000+                    , initialMaxStreamDataBidiRemote = 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) 5000 : plainFrames plain+                }+        | otherwise = plain+    client = do+        waitS+        C.run cc $ \_conn -> threadDelay 1000000++-- | RFC 9000, section 4.1: what the peer has spent is the largest offset+--   it has sent on each stream, gaps and all.  One octet at offset five+--   thousand spends five thousand and one, well past this server's thousand;+--   only the one octet was counted.+testGapCountsAgainstTheWindow+    :: C.ClientConfig -> ServerConfig -> IO () -> IO ()+testGapCountsAgainstTheWindow 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 = StreamF 0 5000 ["x"] False : plainFrames plain}+        | otherwise = plain+    client = do+        waitS+        C.run cc $ \_conn -> threadDelay 1000000++-- | A stream the application has closed, and the peer has not finished, is+--   still held to the limits it was sent under: here a frame for it goes+--   past the limit given as (connection, stream).  Nothing was checked for+--   such a frame.  The server's STOP_SENDING is hidden, so that the client+--   does not reset the stream and end it.+testDepartedOverLimit+    :: (Int, Int) -> C.ClientConfig -> ServerConfig -> IO () -> IO ()+testDepartedOverLimit (connLim, strmLim) cc0 sc0 waitS =+    withAsync server $ \_ -> client `shouldThrow` flowControlError+  where+    sc =+        sc0+            { scParameters =+                (scParameters sc0)+                    { initialMaxData = connLim+                    , initialMaxStreamDataBidiRemote = strmLim+                    }+            , scHooks = (scHooks sc0){onPlainCreated = hide}+            }+    hide _ plain = plain{plainFrames = map unstop $ plainFrames plain}+    unstop StopSending{} = Ping+    unstop f = f+    server = run sc $ \conn -> forever $ do+        strm <- acceptStream conn+        _ <- recvStream strm 1+        closeStream strm+    -- One octet at offset 5000 of stream 0, past whichever limit is the+    -- smaller, along with the data of stream 4, which goes out only once+    -- the server has closed stream 0.+    cc = cc0{ccHooks = (ccHooks cc0){onPlainCreated = inject}}+    inject lvl plain+        | lvl == RTT1Level && any onStream4 (plainFrames plain) =+            plain{plainFrames = StreamF 0 5000 ["x"] False : plainFrames plain}+        | otherwise = plain+    onStream4 (StreamF 4 _ _ _) = True+    onStream4 _ = False+    client = do+        waitS+        C.run cc $ \conn -> do+            s0 <- stream conn+            sendStream s0 "x"+            threadDelay 300000+            s1 <- stream conn+            sendStream s1 "y"+            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@@ -782,6 +1218,23 @@         waitS         C.run cc $ \_conn -> threadDelay 1500000 +-- | Running an action and failing if any thread let an exception go+--   uncaught meanwhile.+--+-- Stopping a server told each connection to close by throwing to its main+-- thread.  'testSlowApp' stops the server while its connection waits for an+-- application that is slow to unwind, and the exception arrived in the+-- teardown, where nothing caught it: the runtime reported an uncaught+-- 'Abort' every time.+noUncaught :: IO () -> IO ()+noUncaught act = do+    caught <- newIORef []+    old <- getUncaughtExceptionHandler+    setUncaughtExceptionHandler $ \e ->+        atomicModifyIORef' caught $ \es -> (show e : es, ())+    act `E.finally` setUncaughtExceptionHandler old+    readIORef caught `shouldReturn` []+ frameEncodingError :: QUICException -> Bool frameEncodingError (TransportErrorIsReceived FrameEncodingError _) = True frameEncodingError _ = False@@ -823,11 +1276,16 @@                 strm <- stream conn                 sendStream strm $ BS.replicate 2048 0                 threadDelay 2000-                sendStream strm $ BS.replicate 2048 0-                shutdownStream strm+                -- The server's STOP_SENDING may have reset the stream by now.+                E.handle stopped $ do+                    sendStream strm $ BS.replicate 2048 0+                    shutdownStream strm                 closeStream strm         r `shouldBe` Just () +    stopped StreamIsClosed = return ()+    stopped e = E.throwIO e+ testCloseWithoutReading :: C.ClientConfig -> ServerConfig -> IO () -> IO () testCloseWithoutReading cc sc0 waitS =     withAsync server $ \_ -> client@@ -842,7 +1300,186 @@         r <- Timeout.timeout 5000000 $ C.run cc $ \conn ->             replicateM_ 20 $ do                 strm <- stream conn-                sendStream strm $ BS.replicate 4096 0-                shutdownStream strm+                -- The server's STOP_SENDING may reset the stream meanwhile.+                E.handle stopped $ do+                    sendStream strm $ BS.replicate 4096 0+                    shutdownStream strm                 closeStream strm         r `shouldBe` Just ()+    stopped StreamIsClosed = return ()+    stopped e = E.throwIO e++-- | Only the closer closed the client's socket, and a connection the peer+--   ends has none: the socket was left open.+testSocketClosed :: C.ClientConfig -> ServerConfig -> IO () -> IO ()+testSocketClosed cc sc waitS = do+    ref <- newIORef Nothing+    withAsync server $ \_ -> do+        waitS+        r <- E.try $ C.run cc $ \conn -> do+            getSocket conn >>= writeIORef ref . Just+            strm <- stream conn+            sendStream strm "ping"+            void $ recvStream strm 1024+        case r of+            Left (ApplicationProtocolErrorIsReceived _ _) -> return ()+            _ -> expectationFailure $ show r+        msock <- readIORef ref+        case msock of+            Nothing -> expectationFailure "no socket"+            Just sock -> NS.withFdSocket sock return `shouldReturn` (-1)+  where+    server = run sc $ \conn -> do+        strm <- acceptStream conn+        _ <- recvStream strm 1024+        abortConnection conn (ApplicationProtocolError 1) "bye"++-- | The server answers a "ping" and returns, which closes the connection+--   with NO_ERROR.  A client waiting on the server then gets+--   'ConnectionIsClosed' at once rather than at the idle timeout, and one+--   not waiting returns what it returns.  'Nothing' expects the former.+--   Either way, the client's socket is closed.+testPeerClosed+    :: C.ClientConfig+    -> ServerConfig+    -> IO ()+    -> Maybe BS.ByteString+    -> (Connection -> IO BS.ByteString)+    -> IO ()+testPeerClosed cc sc waitS expected body = do+    done <- newEmptyMVar+    ref <- newIORef Nothing+    withAsync (server done) $ \_ -> do+        waitS+        r <- Timeout.timeout 3000000 $ E.try $ C.run cc $ \conn -> do+            getSocket conn >>= writeIORef ref . Just+            body conn+        case r of+            Nothing -> expectationFailure "the client waited for the idle timeout"+            Just (Left e) -> case (e, expected) of+                (ConnectionIsClosed _, Nothing) -> return ()+                _ -> expectationFailure $ show e+            Just (Right bs) -> Just bs `shouldBe` expected+        msock <- readIORef ref+        case msock of+            Nothing -> expectationFailure "no socket"+            Just sock -> NS.withFdSocket sock return `shouldReturn` (-1)+        takeMVar done+  where+    server done = run sc $ \conn -> do+        strm <- acceptStream conn+        _ <- recvStream strm 1024+        sendStream strm "pong"+        putMVar done ()++-- | Sending the server a short header packet of each size for a connection+--   it does not know, and the size of the stateless reset it answers with.+resetSizes :: [Int] -> IO [Maybe Int]+resetSizes lens = do+    addr : _ <-+        NS.getAddrInfo+            (Just NS.defaultHints{NS.addrSocketType = NS.Datagram})+            (Just "127.0.0.1")+            (Just "15003")+    E.bracket (NS.openSocket addr) NS.close $ \s -> forM lens $ \len -> do+        _ <-+            NSB.sendTo s (BS.cons 0x40 $ BS.replicate (len - 1) 0x11) $ NS.addrAddress addr+        fmap (BS.length . fst) <$> Timeout.timeout 1000000 (NSB.recvFrom s 2048)++-- | A server that echoes, and tells after each echo where it sent it.+pathEchoServer :: ServerConfig -> MVar NS.SockAddr -> IO ()+pathEchoServer sc paths = run sc $ \conn -> do+    strm <- acceptStream conn+    let loop = do+            bs <- recvStream strm 1024+            unless (bs == "") $ do+                sendStream strm bs+                peerSockAddr <$> getPathInfo conn >>= putMVar paths+                loop+    loop++echo :: Stream -> BS.ByteString -> IO (Maybe BS.ByteString)+echo strm bs = do+    sendStream strm bs+    Timeout.timeout 2000000 $ recvStream strm 1024++-- | RFC 9000 Sec 9.3: only a packet that decrypts moves the peer.  The+--   server took a new address from any packet with its CID on it, before+--   decrypting it, and sent everything there from then on.+testForgedPath+    :: C.ClientConfig+    -> ServerConfig+    -> IO ()+    -> (Version -> CID -> BS.ByteString)+    -> IO ()+testForgedPath cc sc waitS forge = do+    paths <- newEmptyMVar+    withPipe (DropClientPacket []) $+        withAsync (pathEchoServer sc paths) $ \_ -> do+            waitS+            C.run cc $ \conn -> do+                strm <- stream conn+                echo strm "ping1" `shouldReturn` Just "ping1"+                before <- takeMVar paths+                ver <- getVersion conn+                dcid <- getPeerCID conn+                addr : _ <-+                    NS.getAddrInfo+                        (Just NS.defaultHints{NS.addrSocketType = NS.Datagram})+                        (Just "127.0.0.1")+                        (Just "15003")+                E.bracket (NS.openSocket addr) NS.close $ \s ->+                    void $ NSB.sendTo s (forge ver dcid) $ NS.addrAddress addr+                threadDelay 300000+                echo strm "ping2" `shouldReturn` Just "ping2"+                takeMVar paths `shouldReturn` before++-- | A client that moves to a new port is still followed, now that the move+--   is taken from decrypted packets.+testRebindPath :: C.ClientConfig -> ServerConfig -> IO () -> IO ()+testRebindPath cc sc waitS = do+    paths <- newEmptyMVar+    -- Straight to the server: the relay would drop the new port.+    let cc' = cc{C.ccPortName = "15003"}+    withAsync (pathEchoServer sc paths) $ \_ -> do+        waitS+        C.run cc' $ \conn -> do+            strm <- stream conn+            echo strm "ping1" `shouldReturn` Just "ping1"+            before <- takeMVar paths+            void $ controlConnection conn NATRebinding+            -- The first echo may still go out on the old path, while the+            -- new one is being validated.+            let wait (0 :: Int) = expectationFailure "the server never moved"+                wait n = do+                    echo strm "ping" `shouldReturn` Just "ping"+                    p <- takeMVar paths+                    when (p == before) $ do+                        threadDelay 100000+                        wait (n - 1)+            wait 20++-- | RFC 9000 Sec 9.3.2: a packet of the client's, forwarded from another+--   address, moves the server there, and the validation of that address+--   fails.  The server waited for the PATH_RESPONSE for ever and went on+--   sending to the address that never answered.+testSpuriousMigration :: C.ClientConfig -> ServerConfig -> IO () -> IO ()+testSpuriousMigration cc sc waitS = do+    paths <- newEmptyMVar+    trigger <- newEmptyMVar+    withPipe (ForwardAside trigger) $+        withAsync (pathEchoServer sc paths) $ \_ -> do+            waitS+            C.run cc $ \conn -> do+                strm <- stream conn+                echo strm "ping1" `shouldReturn` Just "ping1"+                before <- takeMVar paths+                putMVar trigger ()+                -- Answered on the spurious path, and so not heard until+                -- the server has given up on it.+                sendStream strm "ping2"+                Timeout.timeout 8000000 (recvStream strm 1024)+                    `shouldReturn` Just "ping2"+                _ <- takeMVar paths+                echo strm "ping3" `shouldReturn` Just "ping3"+                takeMVar paths `shouldReturn` before

@@ -4,7 +4,7 @@ import Control.Concurrent.Async (mapConcurrently) import Control.Concurrent.STM import Control.Monad (void)-import Network.Socket (connect, getSocketName)+import Network.Socket (connect, getSocketName, withFdSocket) import Test.Hspec  import Network.QUIC.Internal@@ -44,7 +44,7 @@             (clientConn, _) <- makeConnections serverConf Version1             oldSock <- getSocket clientConn             oldRetired <- getSocketRetire clientConn-            rebind clientConn $ Microseconds 1000000+            rebind clientConn             newSock <- getSocket clientConn             newRetired <- getSocketRetire clientConn             readTVarIO oldRetired `shouldReturn` True@@ -52,6 +52,91 @@             (newSock == oldSock) `shouldBe` False             -- The reader 'rebind' forked is on the new socket; let it go.             atomically $ writeTVar newRetired True++        it "closes the socket it replaced even if the connection ends first" $ do+            (clientConn, _) <- makeConnections serverConf Version1+            oldSock <- getSocket clientConn+            -- A reader on the socket, as 'runClient' has.+            peersa <- peerSockAddr <$> getPathInfo clientConn+            connect oldSock peersa+            retired <- getSocketRetire clientConn+            forkManaged clientConn "QUIC reader" $ readerClient retired oldSock clientConn+            threadDelay 100000+            rebind clientConn+            -- Over at once, torn down as 'runClient' does.+            setDead clientConn+            killReaders clientConn+            freeResources clientConn+            -- A closed socket has its descriptor taken away.+            withFdSocket oldSock return `shouldReturn` (-1)++        it "leaves its socket for the sender when rebind has swapped it out" $ do+            (clientConn, _) <- makeConnections serverConf Version1+            sock <- getSocket clientConn+            peersa <- peerSockAddr <$> getPathInfo clientConn+            connect sock peersa+            retired <- getSocketRetire clientConn+            done <- newEmptyMVar+            void $ forkIO $ readerClient retired sock clientConn >> putMVar done ()+            threadDelay 100000+            -- What 'rebind' does: swap first, retire second.+            newSock <- natRebinding False peersa+            newRetired <- newTVarIO False+            _ <- swapSocket clientConn newSock newRetired+            atomically $ writeTVar retired True+            returned <- timeout (Microseconds 2000000) "retired reader" $ takeMVar done+            returned `shouldBe` Just ()+            -- Left, not closed: the sender could still be inside a send on+            -- it, and it is the sender which closes it.+            void $ getSocketName sock+            closeRetiredSockets clientConn+            getSocketName sock `shouldThrow` anyException+            atomically $ writeTVar newRetired True++        it "leaves its socket even when it is killed rather than told" $ do+            (clientConn, _) <- makeConnections serverConf Version1+            sock <- getSocket clientConn+            peersa <- peerSockAddr <$> getPathInfo clientConn+            connect sock peersa+            retired <- getSocketRetire clientConn+            tid <- forkIO $ readerClient retired sock clientConn+            threadDelay 100000+            newSock <- natRebinding False peersa+            newRetired <- newTVarIO False+            _ <- swapSocket clientConn newSock newRetired+            -- As 'killReaders' does: thrown at, not told.+            killThread tid+            threadDelay 100000+            closeRetiredSockets clientConn+            getSocketName sock `shouldThrow` anyException+            atomically $ writeTVar newRetired True++        it "leaves its socket for the sender when the connection goes idle" $ do+            (clientConn, _) <- makeConnections serverConf Version1+            sock <- getSocket clientConn+            peersa <- peerSockAddr <$> getPathInfo clientConn+            connect sock peersa+            setMinIdleTimeout clientConn $ Microseconds 100000+            retired <- getSocketRetire clientConn+            done <- newEmptyMVar+            void $ forkIO $ readerClient retired sock clientConn >> putMVar done ()+            returned <- timeout (Microseconds 2000000) "idle reader" $ takeMVar done+            returned `shouldBe` Just ()+            void $ getSocketName sock+            closeRetiredSockets clientConn+            getSocketName sock `shouldThrow` anyException++        it "closes a socket it was left even if nothing drains it" $ do+            (clientConn, _) <- makeConnections serverConf Version1+            peersa <- peerSockAddr <$> getPathInfo clientConn+            sock <- natRebinding False peersa+            -- getSocketName says nothing of an unbound socket on Windows.+            connect sock peersa+            retireSocket clientConn sock+            void $ getSocketName sock+            -- No sender to drain it, so the teardown is what closes it.+            freeResources clientConn+            getSocketName sock `shouldThrow` anyException          it "lets a retired reader return, leaving its socket open" $ do             (clientConn, _) <- makeConnections serverConf Version1

@@ -3,11 +3,13 @@ module PacketSpec where  import Control.Monad (forM_)+import Data.Bits (shiftR, xor, (.&.), (.|.)) import Data.ByteString (ByteString) import qualified Data.ByteString as BS import qualified Data.ByteString.Internal as BS import Data.IORef import Data.Tuple (swap)+import Foreign.Ptr (plusPtr) import Test.Hspec  import Network.QUIC.Internal@@ -44,6 +46,15 @@                 [(CryptPacket _ crypt, lvl, _)] <-                     decodeCryptPackets (shortSampleInitial len) True                 decryptCrypt serverConn crypt lvl `shouldReturn` Nothing+        -- RFC 9000 Sec 17.1: a packet number is sent truncated, and is+        -- decoded against the largest received in its own space.+        it "decodes a truncated packet number against its own space" $ do+            (clientConn, serverConn) <- makeConnections serverConf Version1+            setPeerPacketNumber serverConn InitialLevel 299+            bin <- initialWithOneOctetPN clientConn serverConn 300+            (PacketIC (CryptPacket _ crypt) lvl _, _) <- decodePacket bin True+            mplain <- decryptCrypt serverConn crypt lvl+            plainPacketNumber <$> mplain `shouldBe` Just 300     describe "encodePlainPacket" $ do         it "leaves the peer a whole header protection sample" $ do             (senderConn, _) <- makeConnections serverConf Version1@@ -84,6 +95,39 @@             forM_ [0 .. 15] $ \n -> do                 broken <- fst <$> decodePacket (retryWithTrailing n) True                 broken `shouldSatisfy` isBroken++-- | An Initial packet from the client, its packet number in one octet as a+--   peer which has seen @pn - 1@ acknowledged may send it.  Our encoder+--   never truncates that far, so it is put together here.+initialWithOneOctetPN+    :: Connection -> Connection -> PacketNumber -> IO ByteString+initialWithOneOctetPN sender recver pn = do+    let dcid = fromCID clientChosenCID+        payload = BS.cons 0x01 $ BS.replicate 40 0 -- PING, then PADDING+        len = 1 + BS.length payload + 16 -- packet number, payload, tag+        header =+            BS.pack $+                [0xc0, 0x00, 0x00, 0x00, 0x01, fromIntegral (BS.length dcid)]+                    ++ BS.unpack dcid+                    ++ [0x00, 0x00] -- no source CID, no token+                    ++ [0x40 .|. fromIntegral (len `shiftR` 8), fromIntegral len]+                    ++ [fromIntegral pn]+    -- The sample starts four octets past the packet number, which is one+    -- octet.  An AES-GCM coder wants to know where it is before encrypting.+    coder <- getCoder sender InitialLevel False+    sprotector <- getProtector sender InitialLevel+    ciphertext <- BS.createAndTrim (BS.length payload + 64) $ \buf -> do+        setSample sprotector $ buf `plusPtr` 3+        encrypt coder buf payload (AssDat header) pn+    -- The receiver's protector unprotects with the sender's key, and the+    -- mask is the same either way.+    protector <- getProtector recver InitialLevel+    let Mask mask = unprotect protector $ Sample $ BS.take 16 $ BS.drop 3 ciphertext+        flags = BS.head header `xor` (BS.index mask 0 .&. 0x0f)+        epn = BS.last header `xor` BS.index mask 1+    return $+        BS.concat+            [BS.singleton flags, BS.init (BS.tail header), BS.singleton epn, ciphertext]  -- | An Initial packet in a datagram large enough that a server would not --   discard it for being too small, saying its payload is @len@ octets.

@@ -2,6 +2,7 @@  module ReassSpec where +import qualified Data.ByteString as BS import Test.Hspec  import Network.QUIC.Internal@@ -34,12 +35,48 @@                     (\n -> putRxStreamData strm (RxStreamData "x" n 1 False))                     [0 .. fromIntegral maxReassFragments + 100]             map isReassembled answers `shouldSatisfy` and+        -- A copy sent again with other boundaries brings octets of its own,+        -- and taking none of them leaves a gap that is never filled.+        it "takes the new part of a fragment overlapping delivered data" $ do+            strm <- scratchStream serverConf+            putFrag strm 0 10 False `shouldReturn` Reassembled 10+            putFrag strm 5 20 False `shouldReturn` Reassembled 10+            takeRecvStreamQwithSize strm 100 `shouldReturn` octets 0 20+        it "takes data with a FIN when an earlier FIN is held" $ do+            strm <- scratchStream serverConf+            putFrag strm 10 20 True `shouldReturn` Reassembled 10+            putFrag strm 0 20 True `shouldReturn` Reassembled 10+            takeRecvStreamQwithSize strm 100 `shouldReturn` octets 0 20+            takeRecvStreamQwithSize strm 100 `shouldReturn` ""+        it "delivers past a held piece the others cover" $ do+            strm <- scratchStream serverConf+            mapM_+                (\(s, e) -> putFrag strm s e False)+                [(20, 30), (35, 38), (40, 50), (30, 40), (0, 20)]+            takeRecvStreamQwithSize strm 100 `shouldReturn` octets 0 50+        -- Counting a copy twice would charge the connection's window for+        -- octets it already paid for, and the pieces it holds for nothing.+        it "neither counts nor holds a copy of a held fragment" $ do+            strm <- scratchStream serverConf+            putFrag strm 10 20 False `shouldReturn` Reassembled 10+            answers <- mapM (const $ putFrag strm 12 18 False) [0 .. maxReassFragments]+            answers `shouldSatisfy` all (== Duplicated)+            putFrag strm 5 25 False `shouldReturn` Reassembled 10+            putFrag strm 0 5 False `shouldReturn` Reassembled 5+            takeRecvStreamQwithSize strm 100 `shouldReturn` octets 0 25 +-- | The octets from one offset up to another, each its offset modulo 256.+octets :: Int -> Int -> BS.ByteString+octets s e = BS.pack $ map fromIntegral [s .. e - 1]++putFrag :: Stream -> Int -> Int -> Bool -> IO FlowCntl+putFrag strm s e fin = putRxStreamData strm $ RxStreamData (octets s e) s (e - s) fin+ scratchStream :: ServerConfig -> IO Stream scratchStream serverConf = do     (conn, _) <- makeConnections serverConf Version1     newStream conn 0 1000000 1000000  isReassembled :: FlowCntl -> Bool-isReassembled Reassembled = True+isReassembled (Reassembled _) = True isReassembled _ = False

@@ -25,18 +25,28 @@         -- Fifty of them: with one random bit wrong, one reset gets through         -- half the time.  True here is the peer refusing a cleared QUIC Bit,         -- which is what a peer that has not asked for greasing does.-        bss <- mapM (const $ makeStatelessReset token) [1 .. 50 :: Int]+        bss <- mapM (const $ makeStatelessReset 1280 token) [1 .. 50 :: Int]         pkts <- concat <$> mapM (`decodePackets` True) bss         filter broken pkts `shouldBe` []-    it "is 1280 octets, under three times the 428 it answers" $ do-        bs <- makeStatelessReset token-        BS.length bs `shouldBe` 1280-        BS.length bs `shouldSatisfy` (< 3 * 428)+    it "is of the size it is made" $ do+        bs <- makeStatelessReset 21 token+        BS.length bs `shouldBe` 21+    it "is half the packet it answers, up to 1280 octets" $ do+        map statelessResetSizeFor [3000, 2000, 1280, 500, 44]+            `shouldBe` [Just 1280, Just 1000, Just 640, Just 250, Just 22]+    it "is one octet less than a packet of 43 octets or fewer" $ do+        map statelessResetSizeFor [43, 22, 21]+            `shouldBe` [Just 42, Just 21, Nothing]+    it "ends a loop of two endpoints answering each other's resets" $ do+        -- RFC 9000 Sec 10.3.3.  A reset of 1280 octets used to be answered+        -- with another of 1280 octets.+        let rounds = takeWhile (/= Nothing) $ iterate (>>= statelessResetSizeFor) (Just 1500)+        length rounds `shouldSatisfy` (< 40)     it "ends with the token" $ do-        bs <- makeStatelessReset token+        bs <- makeStatelessReset 1280 token         BS.drop (BS.length bs - 16) bs `shouldBe` fromStatelessResetToken token   where-    firstByte = BS.head <$> makeStatelessReset token+    firstByte = BS.head <$> makeStatelessReset 1280 token     token = StatelessResetToken "0123456789abcdef"     broken (PacketIB BrokenPacket _) = True     broken _ = False

@@ -45,7 +45,8 @@             $ \_ -> do                 let cc = addHook cc0 $ setOnPlainCreated largeOffset                 runCnoOp cc ms `shouldThrow` transportErrorsIn [FlowControlError]-        it "MUST send STREAM_LIMIT_ERROR if a stream ID exceeding the limit" $ \_ -> do+        it+            "MUST send STREAM_LIMIT_ERROR if a stream ID exceeds the limit [Transport 4.6]" $ \_ -> do             let cc = addHook cc0 $ setOnPlainCreated largeStreamId             runCnoOp cc ms `shouldThrow` transportErrorsIn [StreamLimitError]         it@@ -60,7 +61,7 @@                         addHook cc0 $ setOnTransportParametersCreated setOriginalDestinationConnectionId                 runCnoOp cc ms `shouldThrow` transportErrorsIn [TransportParameterError]         it-            "MUST send TRANSPORT_PARAMETER_ERROR if preferred_address, is received [Transport 18.2]"+            "MUST send TRANSPORT_PARAMETER_ERROR if preferred_address is received [Transport 18.2]"             $ \_ -> do                 let cc = addHook cc0 $ setOnTransportParametersCreated setPreferredAddress                 runCnoOp cc ms `shouldThrow` transportErrorsIn [TransportParameterError]@@ -85,7 +86,7 @@                 let cc = addHook cc0 $ setOnTransportParametersCreated setMaxUdpPayloadSize                 runCnoOp cc ms `shouldThrow` transportErrorsIn [TransportParameterError]         it-            "MUST send TRANSPORT_PARAMETER_ERROR if ack_delay_exponen > 20 [Transport 7.4 and 18.2]"+            "MUST send TRANSPORT_PARAMETER_ERROR if ack_delay_exponent > 20 [Transport 7.4 and 18.2]"             $ \_ -> do                 let cc = addHook cc0 $ setOnTransportParametersCreated setAckDelayExponent                 runCnoOp cc ms `shouldThrow` transportErrorsIn [TransportParameterError]@@ -119,6 +120,13 @@             $ \_ -> do                 let cc = addHook cc0 $ setOnPlainCreated unknownFrame                 runCnoOp cc ms `shouldThrow` transportErrorsIn [FrameEncodingError]+        it+            "MUST send FRAME_ENCODING_ERROR if a frame is cut short [Transport 12.4]"+            $ \_ -> do+                let cc = addHook cc0 $ setOnPlainCreated truncatedFrame+                -- Not the generic codes 'transportErrorsIn' lets through:+                -- INTERNAL_ERROR is what the decoder's exception used to give.+                runCnoOp cc ms `shouldThrow` frameEncodingError         it "MUST send PROTOCOL_VIOLATION on no frames [Transport 12.4]" $ \_ -> do             let cc = addHook cc0 $ setOnPlainCreated noFrames             runCnoOp cc ms `shouldThrow` transportError@@ -133,7 +141,7 @@                 let cc = addHook cc0 $ setOnPlainCreated cryptoBeyondBuffer                 runCnoOp cc ms `shouldThrow` transportErrorsIn [CryptoBufferExceeded]         it-            "MUST send PROTOCOL_VIOLATION if PATH_CHALLENGE in Handshake is received [Transport 17.2.4]"+            "MUST send PROTOCOL_VIOLATION if PATH_CHALLENGE in Handshake is received [Transport 12.4]"             $ \_ -> do                 let cc = addHook cc0 $ setOnPlainCreated handshakePathChallenge                 runCnoOp cc ms `shouldThrow` transportError@@ -153,7 +161,7 @@                 let cc = addHook cc0 $ setOnPlainCreated handshakeDataBlocked                 runCnoOp cc ms `shouldThrow` transportError         it-            "MUST send PROTOCOL_VIOLATION if reserved bits in Short are non-zero [Transport 17.2]"+            "MUST send PROTOCOL_VIOLATION if reserved bits in Short are non-zero [Transport 17.3.1]"             $ \_ -> do                 let cc = addHook cc0 $ setOnPlainCreated $ rrBits RTT1Level                 runCnoOp cc ms `shouldThrow` transportError@@ -232,15 +240,15 @@                 let cc = cc0{ccALPN = \_ -> return $ Just ["dummy"]}                 runCnoOp cc ms `shouldThrow` cryptoErrorsIn [TLS.NoApplicationProtocol]         it-            "MUST send missing_extension TLS alert if the quic_transport_parameters extension does not included [TLS 8.2]"+            "MUST send missing_extension TLS alert if the quic_transport_parameters extension is not included [TLS 8.2]"             $ \_ -> do                 let cc = addHook cc0 $ setOnTLSExtensionCreated (const [])                 runCnoOp cc ms `shouldThrow` cryptoErrorsIn [TLS.MissingExtension]         it-            "MUST send missing_extension TLS alert if the quic_transport_parameters extension does not included [TLS 8.2]"+            "MUST send missing_extension TLS alert if only the draft quic_transport_parameters extension (0xffa5) is included [TLS 8.2]"             $ \_ -> do                 let f [ExtensionRaw _ v] = [ExtensionRaw (ExtensionID 0xffa5) v]-                    f _ = error "f"+                    f xs = xs                     cc = addHook cc0 $ setOnTLSExtensionCreated f                 runCnoOp cc ms `shouldThrow` cryptoErrorsIn [TLS.MissingExtension]         it@@ -261,9 +269,8 @@                                     }                         runCnoOp cc ms `shouldThrow` transportError                 _ -> do-                    putStrLn-                        "Warning: 0-RTT is not possible. Skipping this test. Use \"h3spec -s 0-RTT\" next time."                     when (ccDebugLog cc0) $ print mres+                    pendingWith "0-RTT is not possible. Use \"h3spec -s 0-RTT\" next time."  ---------------------------------------------------------------- @@ -379,14 +386,14 @@ impossibleAckRange :: EncryptionLevel -> Plain -> Plain impossibleAckRange lvl plain     | lvl == RTT1Level =-        plain{plainFrames = Ack (AckInfo 5 0 [(10, 0)]) 0 : plainFrames plain}+        plain{plainFrames = Ack (AckInfo 5 0 [(10, 0)]) noDelay : plainFrames plain}     | otherwise = plain  -- Nobody has sent a million packets down this connection. ackForUnsentPacket :: EncryptionLevel -> Plain -> Plain ackForUnsentPacket lvl plain     | lvl == RTT1Level =-        plain{plainFrames = Ack (AckInfo 1000000 0 []) 0 : plainFrames plain}+        plain{plainFrames = Ack (AckInfo 1000000 0 []) noDelay : plainFrames plain}     | otherwise = plain  unknownFrame :: EncryptionLevel -> Plain -> Plain@@ -395,6 +402,19 @@         plain{plainFrames = UnknownFrame 0x20 : plainFrames plain}     | otherwise = plain +-- A CRYPTO frame at offset 0 whose length is an eight-octet integer that+-- starts 0x3f: whatever follows, it runs past the end of the packet.  Put+-- together an octet at a time, since no frame this library builds is cut+-- short.+truncatedFrame :: EncryptionLevel -> Plain -> Plain+truncatedFrame lvl plain+    | lvl == RTT1Level =+        plain+            { plainFrames =+                plainFrames plain ++ [UnknownFrame 0x06, UnknownFrame 0x00, UnknownFrame 0xff]+            }+    | otherwise = plain+ -- CRYPTO frames are outside flow control, so nothing but the buffer limit -- stops a peer from parking a fragment far past where the stream has got to -- and having it held.  One octet at this offset is enough to ask for more@@ -562,6 +582,10 @@ -- it detects an error condition; a generic error code (such as -- PROTOCOL_VIOLATION or INTERNAL_ERROR) can always be used in place -- of specific error codes.+frameEncodingError :: QUICException -> Bool+frameEncodingError (TransportErrorIsReceived FrameEncodingError _) = True+frameEncodingError _ = False+ transportErrorsIn :: [TransportError] -> QUICException -> Bool transportErrorsIn tes qe@(TransportErrorIsReceived te _) = (te `elem` tes) || transportError qe transportErrorsIn _ _ = False

@@ -4,8 +4,10 @@  import Control.Concurrent.STM (atomically) import Control.Monad (replicateM)+import qualified Data.ByteString as BS import qualified Data.ByteString.Char8 as C8 import Data.List+import Data.Word (Word8) import qualified System.Timeout as Timeout import Test.Hspec import Test.QuickCheck@@ -21,6 +23,13 @@         1200         RTT1Level +-- | The ACK Delay of an ACK frame whose delay field is the given octets.+ackDelayOf :: [Word8] -> IO Delay+ackDelayOf field = do+    Just [Ack _ d] <-+        decodeFramesBS $ BS.pack $ [0x02, 0x00] ++ field ++ [0x00, 0x00]+    return d+ number :: ReceivedPacket -> Int number rpkt =     let CryptPacket _ crypt = rpCryptPacket rpkt@@ -51,6 +60,28 @@             -- for.             atomically $ prependRecvQ q $ numbered 0             number <$> readRecvQ q `shouldReturn` 0+    -- RFC 9000 Sec 19.3: the field counts units of 2 ^ ack_delay_exponent+    -- microseconds, and it is a varint, so up to 2 ^ 62 - 1 of them.+    describe "ackDelayMicro" $ do+        it "scales the field by the exponent" $ do+            d <- ackDelayOf [0x40, 0x7d] -- 125+            ackDelayMicro 3 d `shouldBe` Microseconds 1000+            ackDelayMicro 0 d `shouldBe` Microseconds 125+        it "saturates instead of overflowing" $ do+            d <- ackDelayOf $ replicate 8 0xff -- 2 ^ 62 - 1+            ackDelayMicro 20 d `shouldBe` Microseconds maxBound+    describe "capAckInfo" $ do+        it "keeps an ACK frame that fits" $ do+            let ai = toAckInfo [9, 8, 7, 5, 4]+            capAckInfo 64 ai `shouldBe` ai+        it "drops the oldest ranges of one that does not" $ do+            let pns = [0, 2 .. 1000] -- 501 ranges+                ai@(AckInfo lpn _ _) = capAckInfo 64 $ toAckInfo $ reverse pns+            bs <- encodeFrames [Ack ai noDelay]+            BS.length bs `shouldSatisfy` (<= 64)+            BS.length bs `shouldSatisfy` (> 60)+            lpn `shouldBe` 1000+            fromAckInfo ai `shouldSatisfy` (`isSuffixOf` pns)     describe "toAckInfo and fromAckInfo" $ do         it "should be dual" $ property $ \xs -> do             let rs =

@@ -0,0 +1,32 @@+module WatchDogSpec where++import Data.IORef+import qualified System.Timeout as Timeout+import Test.Hspec++import Network.QUIC.Internal++spec :: Spec+spec = do+    describe "watchDog" $ do+        it "gives up on a socket it could not open" $ do+            closed <- newIORef False+            r <-+                Timeout.timeout 1000000 $+                    watchDogWith+                        (return (-1))+                        (\_ -> return (-1))+                        (\_ -> writeIORef closed True >> return 0)+                        (return ())+            r `shouldBe` Just ()+            readIORef closed `shouldReturn` False+        it "gives up on a watch that fails" $ do+            -- No descriptor is -1, so the receive fails with EBADF.+            Timeout.timeout 1000000 (watchSocket (-1)) `shouldReturn` Just (-2)+        it "watches again after an interrupted watch" $ do+            n <- newIORef (0 :: Int)+            let watch _ = do+                    i <- atomicModifyIORef' n $ \x -> (x + 1, x)+                    return $ if i < 3 then -1 else -2+            watchDogWith (return 3) watch (\_ -> return 0) (return ())+            readIORef n `shouldReturn` 4