quic 0.3.22 → 0.3.23

raw patch · 29 files changed

+647/−120 lines, 29 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• ChangeLog.md +30/−0
• Network/QUIC.hs +4/−0
• Network/QUIC/Client/Run.hs +2/−2
• Network/QUIC/Config.hs +6/−0
• Network/QUIC/Connection/Crypto.hs +1/−0
• Network/QUIC/Connection/Migration.hs +23/−11
• Network/QUIC/Connection/Stream.hs +42/−10
• Network/QUIC/Connection/Types.hs +3/−2
• Network/QUIC/IO.hs +21/−15
• Network/QUIC/Internal.hs +2/−0
• Network/QUIC/Packet.hs +1/−0
• Network/QUIC/Packet/Encode.hs +83/−8
• Network/QUIC/Receiver.hs +31/−31
• Network/QUIC/Recovery.hs +6/−1
• Network/QUIC/Recovery/Interface.hs +2/−0
• Network/QUIC/Recovery/Release.hs +13/−9
• Network/QUIC/Recovery/Utils.hs +4/−3
• Network/QUIC/Server.hs +1/−0
• Network/QUIC/Server/Reader.hs +50/−7
• Network/QUIC/Server/Run.hs +5/−2
• Network/QUIC/Types/Queue.hs +93/−12
• quic.cabal +1/−1
• test/DatagramSpec.hs +11/−0
• test/HandshakeSpec.hs +41/−0
• test/IOSpec.hs +99/−1
• test/MigrationSpec.hs +22/−1
• test/PacketSpec.hs +2/−2
• test/RecoverySpec.hs +23/−0
• test/TypesSpec.hs +25/−2

@@ -1,5 +1,35 @@ # ChangeLog +## 0.3.23++Adds `scMaxConnections` to `ServerConfig` and `setMaxBidiStreams` and+`setMaxUniStreams` to `Network.QUIC`.  In `Network.QUIC.Internal`,+`newRecvQ` takes a `RecvBudget`, `getMyStreamId` takes a `StreamId`,+`queueTx` and `checkBlocked` are in `STM`, the DATAGRAM queue is a+`DatagramQ`, and `setPeerCIDAndRetireCIDs` is replaced by+`addPeerCIDAndRetire`.++* Take the flow control window in the transaction that finds it+  [#224](https://github.com/kazu-yamamoto/quic/pull/224)+* Add a new peer CID along with the retirement it asks for+  [#225](https://github.com/kazu-yamamoto/quic/pull/225)+* Drop a DATAGRAM when too many wait for the application+  [#227](https://github.com/kazu-yamamoto/quic/pull/227)+* Share one byte budget among a server's receive queues+  [#228](https://github.com/kazu-yamamoto/quic/pull/228)+* Refuse a connection over scMaxConnections with CONNECTION_REFUSED+  [#229](https://github.com/kazu-yamamoto/quic/pull/229)+* Let the application set how many streams the peer may open+  [#230](https://github.com/kazu-yamamoto/quic/pull/230)+* Require a Retry of a client whose token validates nothing+  [#231](https://github.com/kazu-yamamoto/quic/pull/231)+* Check a unidirectional stream against the unidirectional streams we opened+  [#232](https://github.com/kazu-yamamoto/quic/pull/232)+* Answer a frame received in 0-RTT in 1-RTT+  [#233](https://github.com/kazu-yamamoto/quic/pull/233)+* Make no congestion event and send no PING for lost ACK-only packets+  [#234](https://github.com/kazu-yamamoto/quic/pull/234)+ ## 0.3.22  Requires crypto-token 0.2.1.  The public API is unchanged.  In

@@ -36,6 +36,10 @@     resetStream,     stopStream, +    -- ** Limits on the peer's streams+    setMaxBidiStreams,+    setMaxUniStreams,+     -- * IO     recvStream,     resetReceived,

@@ -162,7 +162,7 @@     -- 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+        q <- newRecvQ =<< newRecvBudget maxBound         retire0 <- newTVarIO False         sref <- newIORef (sock, retire0)         pathInfo <- newPathInfo peersa@@ -191,7 +191,7 @@             let myAuthCIDs = defaultAuthCIDs{initSrcCID = Just myCID}                 peerAuthCIDs = defaultAuthCIDs{initSrcCID = Just peerCID, origDstCID = Just peerCID}             genSRT <- makeGenStatelessReset-            connRecvDatagramQ <- newTQueueIO+            connRecvDatagramQ <- newDatagramQ             conn <-                 clientConnection                     conf

@@ -194,6 +194,11 @@     -- ^ Request a certificate from clients.     , scRequireRetry :: Bool     -- ^ Requiring QUIC retry.+    , scMaxConnections :: Int+    -- ^ How many connections may be open at once.  An Initial packet that+    --   would open one more is answered with CONNECTION_REFUSED.  0 or less,+    --   the default being 0, sets no limit; a server open to the Internet+    --   should set one.     , scSessionManager :: SessionManager     -- ^ A session manager of TLS 1.3.     , scDebugLog :: Maybe FilePath@@ -243,6 +248,7 @@         , scALPN = Nothing         , scWantClientCert = False         , scRequireRetry = False+        , scMaxConnections = 0         , scSessionManager = noSessionManager         , scDebugLog = Nothing         , scTicketLifetime = 7200

@@ -19,6 +19,7 @@     initializeCoder,     initializeCoder1RTT,     updateCoder1RTT,+    genNiteCoder,     getCoder,     getProtector,     --

@@ -12,7 +12,7 @@     getNewMyCID,     getMyCIDSeqNum,     setMyCID,-    setPeerCIDAndRetireCIDs,+    addPeerCIDAndRetire,     retirePeerCID,     retireMyCID,     isMyCIDSeqNumIssued,@@ -206,13 +206,26 @@ setPeerRetirePriorTo Connection{..} rpt =     atomically $ modifyTVar' peerCIDDB $ \db -> db{retirePriorTo = rpt} --- | Receiving NewConnectionID-setPeerCIDAndRetireCIDs :: Connection -> Int -> IO [Int]-setPeerCIDAndRetireCIDs Connection{..} rpt = atomically $ do-    db <- readTVar peerCIDDB-    let (db', ns) = arrange rpt db-    writeTVar peerCIDDB db'-    return ns+-- | Receiving NewConnectionID: adding the CID and retiring those below+--   Retire Prior To in one go.  Retired first, nothing might be left to use;+--   added first, the limit is checked only after the retirement+--   (RFC 9000 Sec 5.1.1).  'Nothing' if the limit is exceeded.+addPeerCIDAndRetire :: Connection -> CIDInfo -> Int -> IO (Maybe [Int])+addPeerCIDAndRetire conn@Connection{..} cidInfo rpt = do+    let lim = activeConnectionIdLimit $ getMyParameters conn+    atomically $ do+        db0 <- readTVar peerCIDDB+        let db1+                | Map.member (cidInfoCID cidInfo) (revInfos db0) = db0 -- retransmitted+                | otherwise = add cidInfo db0+            (db2, ns)+                | rpt >= retirePriorTo db0 = arrange rpt db1+                | otherwise = (db1, [])+        if Map.size (revInfos db2) > lim+            then return Nothing+            else do+                writeTVar peerCIDDB db2+                return $ Just ns  arrange :: Int -> CIDDB -> (CIDDB, [Int]) arrange rpt db@CIDDB{..} = (db', dropSeqnums)@@ -220,11 +233,10 @@     (toDrops, cidInfos') = IntMap.partitionWithKey (\k _ -> k < rpt) cidInfos     dropSeqnums = IntMap.foldrWithKey (\k _ ks -> k : ks) [] toDrops     dropCIDs = IntMap.foldr (\c r -> cidInfoCID c : r) [] toDrops-    -- IntMap.findMin is a partial function.-    -- But receiver guarantees that there is at least one cidinfo.+    -- The caller adds a CID at or above rpt first, so one is left.     usedCIDInfo'         | cidInfoSeq usedCIDInfo >= rpt = usedCIDInfo-        | otherwise = snd $ IntMap.findMin cidInfos'+        | otherwise = maybe usedCIDInfo snd $ IntMap.lookupMin cidInfos'     revInfos' = foldr Map.delete revInfos dropCIDs     db' =         db

@@ -14,6 +14,8 @@     updatePeerStreamId,     checkStreamIdRoom,     releasePeerStream,+    setMaxBidiStreams,+    setMaxUniStreams, ) where  import Control.Concurrent.STM@@ -24,12 +26,15 @@ import Network.QUIC.Connection.Types import Network.QUIC.Connector import Network.QUIC.Imports-import Network.QUIC.Parameters import Network.QUIC.Types -getMyStreamId :: Connection -> IO Int-getMyStreamId Connection{..} = do-    next <- currentStream <$> readTVarIO myStreamId+-- | The last stream we opened of the kind 'StreamId' is.+getMyStreamId :: Connection -> StreamId -> IO Int+getMyStreamId Connection{..} sid = do+    let tvar+            | isUnidirectional sid = myUniStreamId+            | otherwise = myStreamId+    next <- currentStream <$> readTVarIO tvar     return $ next - 4  possibleMyStreams :: Connection -> IO Int@@ -149,16 +154,12 @@             | otherwise = peerUniStreamId conn     atomicModifyIORef' ref checkConc   where-    params = getMyParameters conn-    initialStreams-        | dir == Bidirectional = initialMaxStreamsBidi params-        | otherwise = initialMaxStreamsUni params     checkConc conc@Concurrency{..} =         let StreamIdBase base = maxStreams             closed = closedStreams + 1-            base' = initialStreams + closed+            base' = targetStreams + closed             conc' = conc{closedStreams = closed}-         in if base' - base >= max 1 (initialStreams `div` 2)+         in if base' - base >= max 1 (targetStreams `div` 2)                 then (conc'{maxStreams = StreamIdBase base'}, Just base')                 else (conc', Nothing) @@ -182,3 +183,34 @@     dir         | isUnidirectional sid = Unidirectional         | otherwise = Bidirectional++-- | Setting how many bidirectional streams the peer may have open at once,+--   'initialMaxStreamsBidi' to begin with.  A larger number is announced+--   at once.  MAX_STREAMS cannot go down (RFC 9000 Sec 4.6), so a smaller+--   one takes effect as the peer's open streams are closed.+setMaxBidiStreams :: Connection -> Int -> IO ()+setMaxBidiStreams conn = setMaxStreams conn Bidirectional++-- | 'setMaxBidiStreams' for unidirectional streams.+setMaxUniStreams :: Connection -> Int -> IO ()+setMaxUniStreams conn = setMaxStreams conn Unidirectional++setMaxStreams :: Connection -> Direction -> Int -> IO ()+setMaxStreams conn dir n0 = do+    mx <- atomicModifyIORef' ref retarget+    case mx of+        Nothing -> return ()+        Just nms -> sendFrames conn RTT1Level [MaxStreams dir nms]+  where+    ref+        | dir == Bidirectional = peerStreamId conn+        | otherwise = peerUniStreamId conn+    -- RFC 9000 Sec 4.6: no more than 2^60 streams.+    n = max 0 $ min n0 (2 ^ (60 :: Int))+    retarget conc@Concurrency{..} =+        let StreamIdBase base = maxStreams+            base' = min (2 ^ (60 :: Int)) (n + closedStreams)+            conc' = conc{targetStreams = n}+         in if base' > base+                then (conc'{maxStreams = StreamIdBase base'}, Just base')+                else (conc', Nothing)

@@ -195,6 +195,8 @@     , maxStreams :: StreamIdBase     , closedStreams :: Int     -- ^ For the peer's streams: how many of them we are done with+    , targetStreams :: Int+    -- ^ For the peer's streams: how many may be open at once (#86)     }     deriving (Show) @@ -207,6 +209,7 @@         | otherwise = if bidi then 1 else 3     maxStreams = StreamIdBase n     closedStreams = 0+    targetStreams = n  -- | The peer-initiated streams of one type that have been opened so far. --@@ -605,8 +608,6 @@ ----------------------------------------------------------------  type SendStreamQ = TQueue TxStreamData--type DatagramQ = TQueue ByteString  data Shared = Shared     { sharedCloseSent :: IORef Bool

@@ -42,8 +42,8 @@  -- | Counting and queueing at once, so that the offset given is the --   stream's next one and the queue holds a stream's data in order.-queueTx :: Connection -> Stream -> [StreamData] -> Int -> IO ()-queueTx conn s dats len = atomically $ do+queueTx :: Connection -> Stream -> [StreamData] -> Int -> STM ()+queueTx conn s dats len = do     off <- addTxStreamData s len     addTxData conn len     putSendStreamQSTM conn $ TxStreamData s off dats len False@@ -74,16 +74,22 @@     flowControl dats len wait = do         -- FLOW CONTROL: MAX_STREAM_DATA: send: respecting peer's limit         -- FLOW CONTROL: MAX_DATA: send: respecting peer's limit-        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 -> queueTx conn s dats len-                | otherwise -> do-                    let (dats1, dats2) = split n dats+        -- The window is taken in the transaction that finds it.  Two+        -- streams sending at once used to see the same window and send it+        -- twice over.+        eblocked <- atomically $ do+            r <- checkBlocked s len wait+            case r of+                Right n -> do+                    let (dats1, dats2)+                            | len == n = (dats, [])+                            | otherwise = split n dats                     queueTx conn s dats1 n-                    flowControl dats2 (len - n) False+                    return $ Right (n, dats2)+                Left blocked -> return $ Left blocked+        case eblocked of+            Right (n, dats2) ->+                when (n < len) $ flowControl dats2 (len - n) False             Left blocked -> do                 -- Read each time round rather than once: the handshake may                 -- finish while we are waiting for the window, and a BLOCKED@@ -117,8 +123,8 @@       where         len = BS.length bs -checkBlocked :: Stream -> Int -> Bool -> IO (Either Blocked Int)-checkBlocked s len wait = atomically $ do+checkBlocked :: Stream -> Int -> Bool -> STM (Either Blocked Int)+checkBlocked s len wait = do     let conn = streamConnection s     checkPeerClosedSTM conn     strmFlow <- readStreamFlowTx s@@ -343,8 +349,8 @@ -- | Receive all available DATAGRAM frames. --   This function is non-blocking. recvDatagramMany :: Connection -> IO [ByteString]-recvDatagramMany = atomically . flushTQueue . connRecvDatagramQ+recvDatagramMany = atomically . flushDatagramQSTM . connRecvDatagramQ  -- | Receiving a DATAGRAM frame in STM. recvDatagramSTM :: Connection -> STM ByteString-recvDatagramSTM conn = readTQueue (connRecvDatagramQ conn) `orElse` peerClosedSTM conn+recvDatagramSTM conn = readDatagramQSTM (connRecvDatagramQ conn) `orElse` peerClosedSTM conn

@@ -17,6 +17,7 @@     module Network.QUIC.Client.WatchDog,     module Network.QUIC.Socket,     sendOutput,+    processFrame, ) where  import Network.QUIC.Client.Reader (@@ -35,6 +36,7 @@ import Network.QUIC.Parameters import Network.QUIC.QLogger import Network.QUIC.Qlog+import Network.QUIC.Receiver (processFrame) import Network.QUIC.Recovery import Network.QUIC.Sender (sendOutput) import Network.QUIC.Socket

@@ -2,6 +2,7 @@     -- * Encode     encodeVersionNegotiationPacket,     encodeRetryPacket,+    encodeInitialRefusal,     encodePlainPacket,     makeStatelessReset, 

@@ -5,10 +5,13 @@     encodeVersionNegotiationPacket,     encodeRetryPacket,     encodePlainPacket,+    encodeInitialRefusal,     makeStatelessReset, ) where  import qualified Data.ByteString as BS+import qualified Data.ByteString.Short as Short+import Foreign.Marshal.Alloc (allocaBytes) import System.Random (getStdRandom, randomRIO) #if MIN_VERSION_random(1,3,0) import System.Random (uniformByteString)@@ -54,6 +57,36 @@  ---------------------------------------------------------------- +-- | An Initial packet closing with CONNECTION_REFUSED, made without a+--   connection: the keys are the ones the client's Destination Connection+--   ID gives (RFC 9001 Sec 5.2), and that ID is our Source Connection ID.+encodeInitialRefusal :: Version -> CID -> CID -> IO ByteString+encodeInitialRefusal ver peerCID origCID =+    allocaBytes bufLen $ \buf -> allocaBytes bufLen $ \encBuf -> do+        (coder, protector) <-+            genNiteCoder False ver defaultCipher $ initialSecrets ver origCID+        wbuf <- newWriteBuffer buf bufLen+        (epn, epnLen) <-+            encodeLongHeaderPP' True wbuf InitialPacketType ver peerCID origCID (Flags 0) 0+        encodeInt' wbuf 0 -- token+        (len, _) <-+            protectPayloadHeader'+                encBuf+                defaultCipher+                coder+                protector+                wbuf+                [ConnectionClose ConnectionRefused 0 Short.empty]+                0+                epn+                epnLen+                buf+                Nothing+                InitialLevel+        if len < 0 then return BS.empty else BS.packCStringLen (castPtr buf, len)+  where+    bufLen = 1500+ encodeRetryPacket :: RetryPacket -> IO ByteString encodeRetryPacket (RetryPacket ver dCID sCID token (Left odCID)) = withWriteBuffer maximumQUICHeaderSize $ \wbuf -> do     save wbuf@@ -204,9 +237,22 @@     -> PacketNumber     -> IO (EncodedPacketNumber, Int) encodeLongHeaderPP conn wbuf pkttyp ver dCID sCID flags pn = do+    quicBit <- greaseQuicBit <$> getPeerParameters conn+    encodeLongHeaderPP' quicBit wbuf pkttyp ver dCID sCID flags pn++encodeLongHeaderPP'+    :: Bool+    -> WriteBuffer+    -> LongHeaderPacketType+    -> Version+    -> CID+    -> CID+    -> Flags Raw+    -> PacketNumber+    -> IO (EncodedPacketNumber, Int)+encodeLongHeaderPP' quicBit wbuf pkttyp ver dCID sCID flags pn = do     let el@(_, pnLen) = encodePacketNumber 0 {- dummy -} pn         pp = encodePktNumLength pnLen-    quicBit <- greaseQuicBit <$> getPeerParameters conn     Flags flags' <- encodeLongHeaderFlags ver pkttyp flags pp quicBit     write8 wbuf flags'     encodeLongHeader wbuf ver dCID sCID@@ -227,15 +273,46 @@     -> Bool     -> IO (Int, Int) protectPayloadHeader conn wbuf frames pn epn epnLen headerBeg mlen lvl keyPhase = do+    cipher <- getCipher conn lvl+    coder <- getCoder conn lvl keyPhase+    protector <- getProtector conn lvl+    protectPayloadHeader'+        (encodeBuf conn)+        cipher+        coder+        protector+        wbuf+        frames+        pn+        epn+        epnLen+        headerBeg+        mlen+        lvl++-- | 'protectPayloadHeader' with what it takes from a connection given.+protectPayloadHeader'+    :: Buffer+    -> Cipher+    -> Coder+    -> Protector+    -> WriteBuffer+    -> [Frame]+    -> PacketNumber+    -> EncodedPacketNumber+    -> Int+    -> Buffer+    -> Maybe Int+    -> EncryptionLevel+    -> IO (Int, Int)+protectPayloadHeader' encBuf cipher coder protector wbuf frames pn epn epnLen headerBeg mlen lvl = do     -- Real size is maximumUdpPayloadSize. But smaller is better.-    let encBuf = encodeBuf conn-        encBufLen = 1500 - 20 - 8+    let encBufLen = 1500 - 20 - 8     payloadWithoutPaddingSiz <- encodeFramesWithPadding encBuf encBufLen frames-    cipher <- getCipher conn lvl     -- before length or packer number     lengthOrPNBeg <- currentOffset wbuf     (packetLen, headerLen, plainLen, tagLen, padLen) <--        calcLen cipher lengthOrPNBeg payloadWithoutPaddingSiz+        calcLen lengthOrPNBeg payloadWithoutPaddingSiz     when (lvl /= RTT1Level) $ writeLen (epnLen + plainLen + tagLen)     pnBeg <- currentOffset wbuf     writeEpn epnLen@@ -245,8 +322,6 @@     header <- mkBS headerBeg headerLen     --     let sampleBeg = pnBeg `plusPtr` 4-    coder <- getCoder conn lvl keyPhase-    protector <- getProtector conn lvl     setSample protector sampleBeg     len <- encrypt coder cryptoBeg plaintext (AssDat header) pn     --@@ -259,7 +334,7 @@             ok <- withMask protector $ protectHeader headerBeg pnBeg epnLen             return $ if ok then (packetLen, padLen) else (-1, -1)   where-    calcLen cipher lengthOrPNBeg payloadWithoutPaddingSiz = do+    calcLen lengthOrPNBeg payloadWithoutPaddingSiz = do         let headerLen =                 (lengthOrPNBeg `minusPtr` headerBeg)                     -- length: assuming 2byte length

@@ -3,6 +3,7 @@  module Network.QUIC.Receiver (     receiver,+    processFrame, ) where  import qualified Control.Exception as E@@ -11,7 +12,6 @@ import Network.TLS (AlertDescription (..)) import System.Log.FastLogger -import Control.Concurrent.STM import Network.QUIC.Config import Network.QUIC.Connection import Network.QUIC.Connector@@ -259,11 +259,10 @@         ) guardStream _ _ _ = return () --- fixme: what about unidirection stream? streamNotCreatedYet :: Connection -> StreamId -> ReasonPhrase -> IO () streamNotCreatedYet conn sid emsg     | isInitiated conn sid = do-        curSid <- getMyStreamId conn+        curSid <- getMyStreamId conn sid         when (sid > curSid) $             closeConnection conn StreamStateError emsg streamNotCreatedYet _ _ _ = return ()@@ -464,11 +463,18 @@             Reassembled _ -> return ()         deliverStream conn found strm +-- | The level to answer a frame received at: 0-RTT goes from client to+--   server only, so a server answers 0-RTT in 1-RTT (RFC 9000 Sec 17.2.3).+replyLevel :: EncryptionLevel -> EncryptionLevel+replyLevel RTT0Level = RTT1Level+replyLevel lvl = lvl+ processFrame :: Connection -> EncryptionLevel -> Frame -> IO () processFrame _ _ Padding{} = return () processFrame conn lvl Ping = do     -- see ackEli above-    when (lvl /= InitialLevel && lvl /= RTT1Level) $ sendFrames conn lvl []+    when (lvl /= InitialLevel && lvl /= RTT1Level) $+        sendFrames conn (replyLevel lvl) [] processFrame conn lvl (Ack ackInfo@(AckInfo largestAcked _ _) ackDelay) = do     when (lvl == RTT0Level) $ closeConnection conn ProtocolViolation "ACK"     -- RFC 9000 Sec 19.3.1: walking the ranges down from the largest@@ -576,7 +582,7 @@                 -- No more STREAM frames may go out on it.  Without this,                 -- 'sendStream' went on working after the RESET_STREAM.                 finalSize <- resetTxStream strm-                sendFrames conn lvl [ResetStream sid err finalSize]+                sendFrames conn (replyLevel lvl) [ResetStream sid err finalSize] processFrame _ _ (CryptoF _ "") = return () processFrame conn lvl (CryptoF off cdat) = do     when (lvl == RTT0Level) $@@ -704,29 +710,7 @@     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:@@ -736,11 +720,27 @@             -- 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.+            --+            -- Retire Prior To is not above seqNum here, so it does not+            -- increase and is ignored (RFC 9000 Sec 19.15).             sendFrames conn RTT1Level [RetireConnectionID seqNum]         else do-            ng <- not <$> addPeerCID conn cidInfo-            when ng $-                closeConnection conn ConnectionIdLimitError "NEW_CONNECTION_ID limit error"+            -- 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.+            --+            -- The new CID is added along with the retirement, so that one+            -- retiring all the others leaves it to use (#223).+            mseqNums <- addPeerCIDAndRetire conn cidInfo retirePriorTo+            case mseqNums of+                Nothing ->+                    closeConnection conn ConnectionIdLimitError "NEW_CONNECTION_ID limit error"+                Just seqNums ->+                    unless (null seqNums) $+                        sendFrames conn RTT1Level $+                            map RetireConnectionID seqNums 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@@ -805,7 +805,7 @@         closeConnection conn ProtocolViolation "DATAGRAM not supported"     when (1 + lenSize + BS.length dat > limitBytes) $         closeConnection conn ProtocolViolation "DATAGRAM size violation"-    atomically $ writeTQueue (connRecvDatagramQ conn) dat+    writeDatagramQ (connRecvDatagramQ conn) dat processFrame conn _ _ = closeConnection conn ProtocolViolation "Frame is not allowed"  -- Return value indicates duplication.

@@ -31,6 +31,8 @@     -- Release     releaseByRetry,     releaseOldest,+    onPacketsLost,+    retransmit,     -- Timer     beforeAntiAmp,     ldccTimer,@@ -48,6 +50,9 @@     LDCC,     newLDCC,     lostCandidates,+    recoveryCC,+    ptoPing,+    congestionRecoveryStartTime,     SentPackets (..),     qlogSent, ) where@@ -61,4 +66,4 @@ import Network.QUIC.Recovery.Release import Network.QUIC.Recovery.Timer import Network.QUIC.Recovery.Types-import Network.QUIC.Recovery.Utils (sendPing)+import Network.QUIC.Recovery.Utils (retransmit, sendPing)

@@ -46,6 +46,8 @@     unless (null packets) $ do         onPacketsLost ldcc packets         retransmit ldcc packets+        -- Nothing to resend: a PING, to hurry the peer along all the same.+        unless (any spAckEliciting packets) $ sendPing ldcc lvl         setLossDetectionTimer ldcc  resender :: LDCC -> IO ()

@@ -63,16 +63,20 @@ ----------------------------------------------------------------  onPacketsLost :: LDCC -> Seq SentPacket -> IO ()-onPacketsLost ldcc@LDCC{..} lostPackets = case Seq.viewr lostPackets of-    EmptyR -> return ()-    _ :> lastPkt -> do-        decreaseCC ldcc lostPackets-        isRecovery <--            inCongestionRecovery (spTimeSent lastPkt) . congestionRecoveryStartTime-                <$> readTVarIO recoveryCC-        onCongestionEvent ldcc lostPackets isRecovery-        mapM_ (qlogPacketLost ldcc . LostPacket) lostPackets+onPacketsLost ldcc@LDCC{..} lostPackets = do+    decreaseCC ldcc lostPackets+    -- RFC 9002 B.8: only packets in flight make a congestion event.  An+    -- ACK-only packet lost used to halve the window.+    case Seq.viewr inflight of+        EmptyR -> return ()+        _ :> lastPkt -> do+            isRecovery <-+                inCongestionRecovery (spTimeSent lastPkt) . congestionRecoveryStartTime+                    <$> readTVarIO recoveryCC+            onCongestionEvent ldcc inflight isRecovery+    mapM_ (qlogPacketLost ldcc . LostPacket) lostPackets   where+    inflight = Seq.filter ((> 0) . inFlightBytes) lostPackets     onCongestionEvent = updateCC  ----------------------------------------------------------------

@@ -25,10 +25,11 @@  ---------------------------------------------------------------- +-- | Resending the ack-eliciting packets lost.  Lost ACK-only packets need+--   nothing: a PING for them used to send a probe outside the congestion+--   window and resend a packet that was not lost. retransmit :: LDCC -> Seq SentPacket -> IO ()-retransmit ldcc lostPackets-    | null packetsToBeResent = getEncryptionLevel ldcc >>= sendPing ldcc-    | otherwise = mapM_ put packetsToBeResent+retransmit ldcc lostPackets = mapM_ put packetsToBeResent   where     packetsToBeResent = Seq.filter spAckEliciting lostPackets     put = putRetrans ldcc . spPlainPacket

@@ -12,6 +12,7 @@     scALPN,     scWantClientCert,     scRequireRetry,+    scMaxConnections,     scUse0RTT,     scMaxDatagramFrameSize,     scCiphers,

@@ -9,6 +9,7 @@     liveConnections,     spawnConnection,     waitNoConnection,+    releaseConnection,     spawnDispatcher,     waitNoDispatcher,     tokenMgr,@@ -70,6 +71,8 @@     , stopServerNow :: IO ()     -- ^ Stopping this server: what the shutdown handler is handed, and     -- what a connection's 'stop' reaches it by.+    , connectionCount :: TVar Int+    -- ^ Connections forked and not yet ended, for 'scMaxConnections'.     }  newDispatch :: ServerConfig -> IO () -> IO Dispatch@@ -83,6 +86,7 @@         <*> T.newThreadManager         <*> T.newThreadManager         <*> pure stopNow+        <*> newTVarIO 0   where     conf =         CT.defaultConfig@@ -128,6 +132,18 @@ spawnConnection :: Dispatch -> IO () -> IO () spawnConnection Dispatch{..} = T.forkManaged connectionThreads "QUIC connection" +-- | Counting a connection in, unless there are 'scMaxConnections' already.+reserveConnection :: Dispatch -> Int -> IO Bool+reserveConnection Dispatch{..} lim = atomically $ do+    n <- readTVar connectionCount+    if lim <= 0 || n < lim+        then writeTVar connectionCount (n + 1) >> return True+        else return False++-- | Counting a connection out.+releaseConnection :: Dispatch -> IO ()+releaseConnection Dispatch{..} = atomically $ modifyTVar' connectionCount (subtract 1)+ -- | Waiting until no connection is running. waitNoConnection :: Dispatch -> IO () waitNoConnection Dispatch{..} = T.waitUntilAllGone connectionThreads@@ -146,20 +162,24 @@  -- Source CID -> RecvQ -- Initials and RTT0 are queued before Conneciton is created.-newtype RecvQDict = RecvQDict (LRUCacheRef CID RecvQ)+data RecvQDict = RecvQDict RecvBudget (LRUCacheRef CID RecvQ)  recvQDictSize :: Int recvQDictSize = 100 +-- | The queues share one budget, the endpoint's (#170). newRecvQDict :: IO RecvQDict-newRecvQDict = RecvQDict <$> LRUCache.newLRUCacheRef recvQDictSize+newRecvQDict =+    RecvQDict+        <$> newRecvBudget recvBudgetLimit+        <*> LRUCache.newLRUCacheRef recvQDictSize  -- Looking up and insert a new RecvQ if not exist. lookupInsertRecvQDict :: RecvQDict -> CID -> IO (RecvQ, Bool)-lookupInsertRecvQDict (RecvQDict ref) dcid = LRUCache.cached ref dcid newRecvQ+lookupInsertRecvQDict (RecvQDict budget ref) dcid = LRUCache.cached ref dcid $ newRecvQ budget  lookupRecvQDict :: RecvQDict -> CID -> IO (Maybe RecvQ)-lookupRecvQDict (RecvQDict ref) dcid = LRUCache.cached' ref dcid+lookupRecvQDict (RecvQDict _ ref) dcid = LRUCache.cached' ref dcid  ---------------------------------------------------------------- @@ -277,7 +297,7 @@     -> (CryptPacket, EncryptionLevel, Int)     -> IO () dispatch-    Dispatch{..}+    d@Dispatch{..}     ServerConfig{..}     forkConnection     logAction@@ -330,7 +350,9 @@                                 -- Retry -- but the address it arrives from has                                 -- proved nothing.                                 fresh <- isTokenFresh ct-                                pushToAcceptFirst $ fresh && isTokenAddress ct peersa+                                if fresh && isTokenAddress ct peersa+                                    then pushToAcceptFirst True+                                    else unvalidated                         -- A token we cannot decrypt is not a token.  RFC 9000                         -- section 8.1.3: "If the token is invalid, then the                         -- server SHOULD proceed as if the client did not have@@ -339,14 +361,35 @@                         -- rubbish than for sending nothing, and turned off                         -- the anti-amplification limit for an address that                         -- had proved nothing.-                        Nothing -> pushToAcceptFirst False+                        Nothing -> unvalidated                 Just conn -> writeRecvQ (connRecvQ conn) $ mkReceivedPacket cpkt tim siz lvl       where         myVersions = scVersions+        -- An address no token has validated: a Retry if one is required,+        -- as for an Initial with no token at all.+        unvalidated+            | scRequireRetry = sendRetry+            | otherwise = pushToAcceptFirst False         pushToAcceptQ myAuthCIDs peerAuthCIDs addrValid = do             let key = nonZeroLengthCID sCID peersa+            mq <- lookupRecvQDict srcTable key+            case mq of+                -- A retransmission of an Initial already accepted.+                Just q -> writeRecvQ q $ mkReceivedPacket cpkt tim siz lvl+                Nothing -> do+                    ok <- reserveConnection d scMaxConnections+                    if ok+                        then accept myAuthCIDs peerAuthCIDs addrValid key+                        else -- RFC 9000 Sec 5.2.2: "If a server refuses to+                        -- accept a new connection, it SHOULD send an Initial+                        -- packet containing a CONNECTION_CLOSE frame with+                        -- error code CONNECTION_REFUSED."+                            encodeInitialRefusal peerVer sCID dCID >>= send'+        accept myAuthCIDs peerAuthCIDs addrValid key = do             (q, exist) <- lookupInsertRecvQDict srcTable key             writeRecvQ q $ mkReceivedPacket cpkt tim siz lvl+            -- Another dispatcher got there first.+            when exist $ releaseConnection d             unless exist $ do                 let reg = registerConnectionDict dstTable                     unreg cid =

@@ -234,7 +234,7 @@     -> Dispatch     -> Accept     -> IO ()-runServer conf server0 dispatch acc = do+runServer conf server0 dispatch acc = (`E.finally` done) $ do     labelMe "QUIC runServer"     E.bracket open clse $ \(ConnRes conn myAuthCIDs _reader) ->         handleLogUnit (debugLog conn) $ do@@ -316,6 +316,9 @@             setConnectionClosed conn             closure conn ldcc ex   where+    done = do+        closeRecvQ $ accRecvQ acc+        releaseConnection dispatch     open = createServerConnection conf dispatch acc     -- 'killReaders' stops this connection's own threads and sees them out,     -- which has to happen before the buffers they use are freed.  Nothing@@ -367,7 +370,7 @@     (debugLog, dclean) <- dirDebugLogger scDebugLog ocid `E.onException` qclean     flip E.onException (dclean >> qclean) $ do         debugLog $ "Original CID: " <> bhow ocid-        connRecvDatagramQ <- newTQueueIO+        connRecvDatagramQ <- newDatagramQ         conn <-             serverConnection                 conf

@@ -1,7 +1,8 @@ module Network.QUIC.Types.Queue where  import Control.Concurrent.STM-import Control.Monad (when)+import Control.Monad (void, when)+import Data.ByteString (ByteString)  import Network.QUIC.Types.Packet @@ -23,18 +24,44 @@ recvQLimit :: Int recvQLimit = 4096 -data RecvQ = RecvQ (TQueue ReceivedPacket) (TVar Int)+-- | How many bytes the packets waiting in all of a server's 'RecvQ's may+--   take together.  'recvQLimit' bounds one connection, and a server has as+--   many connections as it is asked for, so this is what bounds the server+--   (#170).  Some ten connections' worth of full queues.+recvBudgetLimit :: Int+recvBudgetLimit = 64 * 1024 * 1024 -newRecvQ :: IO RecvQ-newRecvQ = RecvQ <$> newTQueueIO <*> newTVarIO 0+-- | Bytes in use, shared by the queues of an endpoint, and its limit.+data RecvBudget = RecvBudget Int (TVar Int) +newRecvBudget :: Int -> IO RecvBudget+newRecvBudget lim = RecvBudget lim <$> newTVarIO 0++-- | Packets, how many, their bytes, whether closed, and the budget.+data RecvQ+    = RecvQ (TQueue ReceivedPacket) (TVar Int) (TVar Int) (TVar Bool) RecvBudget++newRecvQ :: RecvBudget -> IO RecvQ+newRecvQ budget =+    RecvQ+        <$> newTQueueIO+        <*> newTVarIO 0+        <*> newTVarIO 0+        <*> newTVarIO False+        <*> pure budget+ readRecvQ :: RecvQ -> IO ReceivedPacket-readRecvQ (RecvQ q qlen) = atomically $ do+readRecvQ (RecvQ q qlen qbytes _ (RecvBudget _ used)) = atomically $ do     x <- readTQueue q+    let n = rpReceivedBytes x     modifyTVar' qlen (subtract 1)+    modifyTVar' qbytes (subtract n)+    modifyTVar' used (subtract n)     return x --- | Queue a packet, or drop it if 'recvQLimit' are already waiting.+-- | Queue a packet, or drop it if 'recvQLimit' are already waiting, if the+--   budget it shares with the other queues has no room for it, or if the+--   queue has been closed. -- -- What is dropped is the packet that has just arrived rather than one -- already queued, so that what the connection has taken on is kept and only@@ -45,18 +72,72 @@ -- is one sending faster than the receiver can refuse what it sends, and a -- line of log for each of those packets is the same attack by another road. writeRecvQ :: RecvQ -> ReceivedPacket -> IO ()-writeRecvQ (RecvQ q qlen) x = atomically $ do+writeRecvQ (RecvQ q qlen qbytes closed (RecvBudget lim used)) x = atomically $ do+    c <- readTVar closed     n <- readTVar qlen-    when (n < recvQLimit) $ do+    u <- readTVar used+    let b = rpReceivedBytes x+    when (not c && n < recvQLimit && u + b <= lim) $ do         writeTQueue q x         writeTVar qlen $! n + 1+        modifyTVar' qbytes (+ b)+        writeTVar used $! u + b  -- | Put back a packet this connection had already taken. ----- Not subject to the limit: these come from the packets put off for want of--- the keys to read them, which the handshake bounds, and dropping one would--- lose a packet the connection has already accounted for.+-- Not subject to the limits: these come from the packets put off for want+-- of the keys to read them, which the handshake bounds, and dropping one+-- would lose a packet the connection has already accounted for. prependRecvQ :: RecvQ -> ReceivedPacket -> STM ()-prependRecvQ (RecvQ q qlen) x = do+prependRecvQ (RecvQ q qlen qbytes _ (RecvBudget _ used)) x = do+    let b = rpReceivedBytes x     unGetTQueue q x     modifyTVar' qlen (+ 1)+    modifyTVar' qbytes (+ b)+    modifyTVar' used (+ b)++-- | Closing a queue nobody will read again: what it holds goes back to the+--   budget, and what is written to it from now on is dropped.  A queue the+--   dispatcher can still find, for a connection that has ended, would+--   otherwise hold its share of the budget for good.+closeRecvQ :: RecvQ -> IO ()+closeRecvQ (RecvQ q qlen qbytes closed (RecvBudget _ used)) = atomically $ do+    writeTVar closed True+    b <- readTVar qbytes+    modifyTVar' used (subtract b)+    writeTVar qbytes 0+    writeTVar qlen 0+    void $ flushTQueue q++----------------------------------------------------------------++-- | How many DATAGRAM frames a connection may have waiting for the+--   application.  Each fits in a packet, so this is a megabyte and a half+--   or so (#226).+datagramQLimit :: Int+datagramQLimit = 1024++data DatagramQ = DatagramQ (TQueue ByteString) (TVar Int)++newDatagramQ :: IO DatagramQ+newDatagramQ = DatagramQ <$> newTQueueIO <*> newTVarIO 0++readDatagramQSTM :: DatagramQ -> STM ByteString+readDatagramQSTM (DatagramQ q qlen) = do+    x <- readTQueue q+    modifyTVar' qlen (subtract 1)+    return x++flushDatagramQSTM :: DatagramQ -> STM [ByteString]+flushDatagramQSTM (DatagramQ q qlen) = do+    writeTVar qlen 0+    flushTQueue q++-- | Queue a DATAGRAM, or drop it if 'datagramQLimit' are already waiting.+--   RFC 9221 Sec 5 lets a receiver drop DATAGRAM frames.+writeDatagramQ :: DatagramQ -> ByteString -> IO ()+writeDatagramQ (DatagramQ q qlen) x = atomically $ do+    n <- readTVar qlen+    when (n < datagramQLimit) $ do+        writeTQueue q x+        writeTVar qlen $! n + 1

@@ -1,6 +1,6 @@ cabal-version:      2.0 name:               quic-version:            0.3.22+version:            0.3.23 license:            BSD3 license-file:       LICENSE maintainer:         kazu@iij.ad.jp

@@ -3,6 +3,7 @@ module DatagramSpec where  import Control.Concurrent+import Control.Concurrent.STM (atomically) import qualified Control.Exception as E import Control.Monad import Data.ByteString ()@@ -138,6 +139,16 @@                         sendDatagram conn (BS.replicate 50 0x41)                         threadDelay 500 +    -- #226: the queue grew for as long as the peer sent faster than the+    -- application read.+    describe "the DATAGRAM receive queue" $+        it "drops what arrives past its limit" $ do+            q <- newDatagramQ+            forM_ [1 .. datagramQLimit + 10] $ \_ -> writeDatagramQ q "x"+            length <$> atomically (flushDatagramQSTM q)+                `shouldReturn` datagramQLimit+            writeDatagramQ q "y"+            atomically (flushDatagramQSTM q) `shouldReturn` ["y"]     describe "Datagram SHOULD requirements" $ do         -- "This frame SHOULD be sent as soon as possible... and MAY be coalesced with other frames."         it "Can send and receive DATAGRAM in 1-RTT" $ do

@@ -4,6 +4,7 @@ module HandshakeSpec where  import Control.Concurrent+import Control.Concurrent.Async (wait, withAsync) import qualified Control.Exception as E import Control.Monad import qualified Data.ByteString as BS@@ -78,6 +79,21 @@             let cc = testClientConfig                 sc = sc0{scRequireRetry = True}             testHandshake cc sc waitS FullHandshake+        -- A token that proves nothing does not stand in for the Retry.+        it "requires a Retry of a client with an invalid token" $ do+            let cc =+                    testClientConfig+                        { ccResumption =+                            (ccResumption testClientConfig)+                                { resumptionToken = "invalid token"+                                , resumptionVersion = ccVersion testClientConfig+                                }+                        }+                sc = sc0{scRequireRetry = True}+            E.bracket (forkIO $ S.run sc $ \_ -> threadDelay 2000000) killThread $ \_ -> do+                waitS+                C.run cc (\conn -> waitEstablished conn >> retry <$> getConnectionInfo conn)+                    `shouldReturn` True         it "can handshake in the case of resumption" $ do             let cc = testClientConfig                 sc = sc0@@ -152,6 +168,31 @@                         te == cryptoError TLS.HandshakeFailure                     | otherwise = False             testHandshake3 cc1 cc2 sc waitS handshakeFailure+        -- RFC 9000 Sec 5.2.2: a connection over the limit is refused with+        -- CONNECTION_REFUSED, and one ending makes room for another (#170).+        it "refuses a connection over scMaxConnections" $ do+            let sc = sc0{scMaxConnections = 1}+            first <- newEmptyMVar+            done <- newEmptyMVar+            let holder = C.run testClientConfig $ \conn -> do+                    waitEstablished conn+                    putMVar first ()+                    takeMVar done+                refused (TransportErrorIsReceived ConnectionRefused _) = True+                refused _ = False+                establish = C.run testClientConfig waitEstablished+                again 0 = establish+                again n =+                    establish `E.catch` \(_ :: QUICException) ->+                        threadDelay 200000 >> again (n - 1 :: Int)+            E.bracket (forkIO $ S.run sc $ \_ -> threadDelay 10000000) killThread $ \_ -> do+                waitS+                withAsync holder $ \h -> do+                    takeMVar first+                    establish `shouldThrow` refused+                    putMVar done ()+                    wait h+                again 10         it "reconnects with the version a Version Negotiation offers" $ do             let cc = testClientConfig{ccVersion = Version2, ccVersions = [Version2, Version1]}                 sc = sc0{scVersions = [Version1]}

@@ -6,7 +6,7 @@  import Control.Concurrent import Control.Concurrent.Async-import Control.Concurrent.STM (readTVarIO)+import Control.Concurrent.STM (atomically, readTVarIO) import qualified Control.Exception as E import Control.Monad import Data.Bits (shiftR)@@ -15,6 +15,7 @@ import Data.IORef import qualified Data.IntMap.Strict as IntMap import GHC.Conc (getUncaughtExceptionHandler, setUncaughtExceptionHandler)+import Network.Control (txfSent) import qualified Network.Socket as NS import qualified Network.Socket.ByteString as NSB import System.Directory (listDirectory)@@ -79,6 +80,24 @@             -- The first one taken and lost: the second keeps its offset.             _ <- takeSendStreamQ clientConn             offset <$> takeSendStreamQ clientConn `shouldReturn` (6, 16)+    -- RFC 9000 Sec 4.1: two streams sending at once used to both see the+    -- same connection window and queue twice what MAX_DATA allowed.+    describe "the connection window" $+        it "is not given to two streams sending at once" $ do+            scw <- makeTestServerConfig+            (clientConn, _) <- makeConnections scw Version1+            overs <- forM [0 .. 999] $ \i -> do+                s1 <- addStream clientConn (8 * i)+                s2 <- addStream clientConn (8 * i + 4)+                setTxMaxStreamData s1 100+                setTxMaxStreamData s2 100+                base <- txfSent <$> atomically (readConnectionFlowTx clientConn)+                setTxMaxData clientConn $ base + 100+                let send s = Timeout.timeout 1000 $ sendStream s $ BS.replicate 100 0+                _ <- concurrently (send s1) (send s2)+                sent <- txfSent <$> atomically (readConnectionFlowTx clientConn)+                return $ sent - base+            filter (> 100) overs `shouldBe` []     describe "a stateless reset to the client" $         it "ends the connection with ConnectionIsReset" $             testStatelessResetReceived cc sc waitS@@ -173,11 +192,48 @@             withPipe (DelayServerPacket 300) $ testLateCopy False cc sc waitS         it "ignores a late copy of the data it received on a stream the peer opened" $ do             withPipe (DelayClientPacket 300) $ testLateCopy True cc sc waitS+    -- RFC 9000 Sec 17.2.3: 0-RTT goes from client to server only.+    describe "a frame received in 0-RTT" $ do+        let replied serverConn = do+                out <- atomically $ takeOutputSTM serverConn+                case out of+                    OutControl lvl frames -> return (lvl, frames)+                    _ -> fail "no control frames"+        it "has its ACK sent in 1-RTT" $ do+            scz <- makeTestServerConfig+            (_, serverConn) <- makeConnections scz Version1+            processFrame serverConn RTT0Level Ping+            fst <$> replied serverConn `shouldReturn` RTT1Level+        it "has its RESET_STREAM sent in 1-RTT" $ do+            scz <- makeTestServerConfig+            (_, serverConn) <- makeConnections scz Version1+            _ <- addStream serverConn 0+            processFrame serverConn RTT0Level $ StopSending 0 $ ApplicationProtocolError 0+            (lvl, frames) <- replied serverConn+            lvl `shouldBe` RTT1Level+            frames `shouldSatisfy` any isResetStream     describe "stream limits" $ do+        -- #86: the number the peer may have open is no longer fixed.+        it "announces a raised limit at once and holds a lowered one" $ do+            scl <- makeTestServerConfig+            (_, serverConn) <- makeConnections scl Version1+            let initial = initialMaxStreamsBidi $ getMyParameters serverConn+            setMaxBidiStreams serverConn $ initial + 10+            out <- atomically $ takeOutputSTM serverConn+            case out of+                OutControl RTT1Level [MaxStreams Bidirectional n] -> n `shouldBe` initial + 10+                _ -> expectationFailure "no MAX_STREAMS"+            -- MAX_STREAMS cannot go down, and closing one stream of many+            -- under a lower target gives the peer no new one.+            setMaxBidiStreams serverConn 1+            atomically (isEmptyOutputSTM serverConn) `shouldReturn` True+            checkStreamIdRoom serverConn Bidirectional `shouldReturn` Nothing         it "keeps the peer's open bidirectional streams within the limit" $ do             withPipe (DropClientPacket []) $ testOpenStreams cc sc waitS         it "raises the limit on unidirectional streams by their own number" $ do             withPipe (DropClientPacket []) $ testUniStreams cc sc waitS+        it "ignores STOP_SENDING for a unidirectional stream it has closed" $ do+            withPipe (DropClientPacket []) $ testStopClosedUni 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@@ -659,6 +715,48 @@         forever $ do             strm <- acceptStream conn             recvStream strm 1++isResetStream :: Frame -> Bool+isResetStream ResetStream{} = True+isResetStream _ = False++-- | A unidirectional stream used to be checked against the counter of+--   bidirectional ones, so a client that had opened none of those closed the+--   connection with STREAM_STATE_ERROR on a STOP_SENDING for stream 2.+testStopClosedUni :: C.ClientConfig -> ServerConfig -> IO () -> IO ()+testStopClosedUni cc sc waitS = do+    closed <- newEmptyMVar+    stopped <- newEmptyMVar+    withAsync (server closed stopped) $ \_ -> client closed stopped+  where+    client closed stopped = do+        waitS+        C.run cc $ \conn -> do+            uni <- unidirectionalStream conn+            sendStream uni "x"+            closeStream uni+            putMVar closed ()+            () <- takeMVar stopped+            threadDelay 200000+            strm <- stream conn+            sendStream strm "ping"+            shutdownStream strm+            recvStream strm 4 `shouldReturn` "pong"+    server closed stopped = run sc $ \conn -> do+        uni <- acceptStream conn+        recvStream uni 1 `shouldReturn` "x"+        () <- takeMVar closed+        -- Sent as is: 'stopStream' sends nothing for a stream whose FIN+        -- has arrived, and a peer's may cross it.+        sendFrames+            conn+            RTT1Level+            [StopSending (streamId uni) $ ApplicationProtocolError 0]+        putMVar stopped ()+        strm <- acceptStream conn+        _ <- recvStream strm 4+        sendStream strm "pong"+        shutdownStream strm  testRecvStreamClientStopFirst     :: C.ClientConfig -> ServerConfig -> IO () -> IO ()

@@ -5,7 +5,7 @@ import Control.Concurrent import Control.Concurrent.Async (concurrently, mapConcurrently, wait, withAsync) import Control.Concurrent.STM-import Control.Monad (forM, void)+import Control.Monad (forM, forM_, void) import qualified Data.ByteString as BS import qualified Data.ByteString.Short as Short import Network.Control (txfLimit)@@ -51,6 +51,27 @@     -- CID of its own on the new path, while the old path keeps the old     -- one; the old one was given up at once, and the old path's challenge     -- went out with the new.+    -- #223: a NEW_CONNECTION_ID retiring every CID it does not bring+    -- left none to use, and the connection ended with INTERNAL_ERROR.+    describe "addPeerCIDAndRetire" $ do+        let cid :: Int -> CID+            cid i = toCID $ BS.replicate 8 $ fromIntegral i+            srt i = StatelessResetToken $ Short.toShort $ BS.replicate 16 $ fromIntegral i+            info i = newCIDInfo i (cid i) (srt i)+        it "moves to the new CID when all the others are retired" $ do+            (_, serverConn) <- makeConnections serverConf Version1+            addPeerCIDAndRetire serverConn (info 1) 1 `shouldReturn` Just [0]+            getPeerCID serverConn `shouldReturn` cid 1+        it "lets a CID exceed the limit when the excess is retired" $ do+            (_, serverConn) <- makeConnections serverConf Version1+            let lim = activeConnectionIdLimit $ getMyParameters serverConn+            forM_ [1 .. lim - 1] $ \i ->+                addPeerCIDAndRetire serverConn (info i) 0 `shouldReturn` Just []+            addPeerCIDAndRetire serverConn (info lim) 0 `shouldReturn` Nothing+            addPeerCIDAndRetire serverConn (info lim) lim+                `shouldReturn` Just [0 .. lim - 1]+            getPeerCID serverConn `shouldReturn` cid lim+     describe "validateMigration" $ do         let next conn = do                 out <- T.timeout 4000000 $ atomically $ takeOutputSTM conn

@@ -180,7 +180,7 @@     -- dummy     let clientConf = testClientConfig     (sock, peersa) <- clientSocket "127.0.0.1" "2000"-    q <- newRecvQ+    q <- newRecvQ =<< newRecvBudget maxBound     retire <- newTVarIO False     sref <- newIORef (sock, retire)     pathInfo <- newPathInfo peersa@@ -188,7 +188,7 @@     let ver = v         verInfo = VersionInfo ver [ver]     genSRT <- makeGenStatelessReset-    dq <- newTQueueIO+    dq <- newDatagramQ     ----     clientConn <-         clientConnection

@@ -4,6 +4,7 @@  import Control.Concurrent.STM import Data.ByteString.Short ()+import Data.Maybe (isJust) import qualified Data.Sequence as Seq import Data.UnixTime import Test.Hspec@@ -51,6 +52,28 @@             atomically $ writeTVar (encryptionLevel cs) RTT1Level             onAckReceived ldcc HandshakeLevel (AckInfo 0 0 []) 0             getPtoTimeAndSpace ldcc `shouldReturn` Nothing+    -- RFC 9002 B.8: only packets in flight make a congestion event.+    describe "a lost packet" $ do+        let lostWith frames = do+                cs <- newConnState Client+                ldcc <- newLDCC cs (\_ -> return ()) (\_ -> return ())+                now <- getTimeMicrosecond+                let lost =+                        (sp 1 (any ackEliciting frames) now)+                            { spEncryptionLevel = RTT1Level+                            , spSentBytes = 1200+                            , spPlainPacket = PlainPacket (Short $ toCID "") (Plain (Flags 0) 1 frames 0)+                            }+                onPacketsLost ldcc $ Seq.singleton lost+                retransmit ldcc $ Seq.singleton lost+                -- A congestion event starts a recovery period.+                event <- isJust . congestionRecoveryStartTime <$> readTVarIO (recoveryCC ldcc)+                ping <- readTVarIO $ ptoPing ldcc+                return (event, ping)+        it "is no congestion event and needs no PING if it carried only an ACK" $+            lostWith [Ack (AckInfo 0 0 []) noDelay] `shouldReturn` (False, Nothing)+        it "is a congestion event if it was in flight" $+            fst <$> lostWith [Ping] `shouldReturn` True     describe "persistent congestion" $ do         it "does not find a pair" $ do             findDuration (Seq.fromList [sp0]) 0 `shouldBe` Nothing

@@ -43,7 +43,7 @@     -- anyone who can see a connection ID can fill it.  See #167.     describe "RecvQ" $ do         it "keeps what it has taken and drops what is past the limit" $ do-            q <- newRecvQ+            q <- newRecvQ =<< newRecvBudget maxBound             mapM_ (writeRecvQ q . numbered) [1 .. recvQLimit + 100]             ns <- replicateM recvQLimit (number <$> readRecvQ q)             -- The packets kept are the ones that arrived first: what is@@ -52,7 +52,7 @@             ns `shouldBe` [1 .. recvQLimit]             Timeout.timeout 100000 (readRecvQ q) `shouldReturn` Nothing         it "takes back a packet it had already taken, full or not" $ do-            q <- newRecvQ+            q <- newRecvQ =<< newRecvBudget maxBound             mapM_ (writeRecvQ q . numbered) [1 .. recvQLimit]             -- Putting back what was put off for want of keys is not the             -- same as taking something new on, and the limit does not apply@@ -60,6 +60,29 @@             -- for.             atomically $ prependRecvQ q $ numbered 0             number <$> readRecvQ q `shouldReturn` 0+        -- #170: what one connection may queue times the number of+        -- connections was a limit on nothing.+        it "shares a budget with the other queues of the endpoint" $ do+            budget <- newRecvBudget 3600 -- three packets of 1200+            q1 <- newRecvQ budget+            q2 <- newRecvQ budget+            mapM_ (writeRecvQ q1 . numbered) [1, 2]+            mapM_ (writeRecvQ q2 . numbered) [3, 4]+            number <$> readRecvQ q1 `shouldReturn` 1+            writeRecvQ q2 $ numbered 5+            number <$> readRecvQ q2 `shouldReturn` 3+            number <$> readRecvQ q2 `shouldReturn` 5+        it "gives back what a closed queue holds and takes no more" $ do+            budget <- newRecvBudget 2400+            q1 <- newRecvQ budget+            q2 <- newRecvQ budget+            mapM_ (writeRecvQ q1 . numbered) [1, 2]+            closeRecvQ q1+            writeRecvQ q1 $ numbered 3+            mapM_ (writeRecvQ q2 . numbered) [4, 5]+            number <$> readRecvQ q2 `shouldReturn` 4+            number <$> readRecvQ q2 `shouldReturn` 5+            Timeout.timeout 100000 (readRecvQ q1) `shouldReturn` Nothing     -- 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