gbnet-hs 0.2.1.0 → 0.2.2.0
raw patch · 21 files changed
+654/−447 lines, 21 filesPVP ok
version bump matches the API change (PVP)
API changes (from Hackage documentation)
Files
- CHANGELOG.md +14/−0
- README.md +25/−19
- gbnet-hs.cabal +1/−1
- src/GBNet/Channel.hs +17/−19
- src/GBNet/Congestion.hs +19/−19
- src/GBNet/Connection.hs +46/−48
- src/GBNet/Fragment.hs +8/−8
- src/GBNet/Net.hs +2/−2
- src/GBNet/Peer.hs +118/−117
- src/GBNet/Peer/Handshake.hs +18/−16
- src/GBNet/Peer/Internal.hs +4/−3
- src/GBNet/Reliability.hs +29/−29
- src/GBNet/Replication/Delta.hs +25/−24
- src/GBNet/Replication/Interpolation.hs +8/−8
- src/GBNet/Replication/Priority.hs +3/−3
- src/GBNet/Security.hs +7/−9
- src/GBNet/Socket.hs +13/−11
- src/GBNet/TestNet.hs +9/−10
- src/GBNet/Util.hs +6/−1
- src/GBNet/ZeroCopy.hs +4/−4
- test/Main.hs +278/−96
CHANGELOG.md view
@@ -1,5 +1,19 @@ # Changelog +## 0.2.2.0++### Bug Fixes++- **Bandwidth tracking now functional**: `recordBytesSent` and `recordBytesReceived` were implemented in `Connection` but never called. Wired into `peerTick`: outgoing bytes recorded after encryption in `drainAllConnectionQueues`, incoming bytes recorded per-packet in `handlePacket`. `BandwidthTracker` and `NetworkStats` byte counters now report real values.+- **Fix O(n^2) list appending** in `encryptOutgoing` and `processPacketsPure`: replaced `acc ++ [x]` with reverse-accumulator pattern.+- **Fix migration cooldowns leak**: `npMigrationCooldowns` map now swept of stale entries in `updateConnections`, preventing unbounded growth on long-running servers.+- **Total function compliance**: replaced partial `Seq.index` with `Seq.lookup` in delta encoding/baseline management.++### Internal++- Codebase-wide elimination of prime-mark variable naming (`x'`, `x''`) in favour of descriptive names and optics composition across all 31 source files, tests, and benchmarks.+- README and EXAMPLES: removed all prime-mark variables, added optics-based configuration examples, fixed section numbering, added `LambdaCase` pragma.+ ## 0.2.1.0 ### Bug Fixes
README.md view
@@ -43,6 +43,7 @@ ### Simple Game Loop ```haskell+{-# LANGUAGE LambdaCase #-} import GBNet import Control.Monad.IO.Class (liftIO) @@ -63,12 +64,12 @@ gameLoop peer = do -- Single call: receive, process, broadcast, send let outgoing = [(ChannelId 0, encodeMyState myState)]- (events, peer') <- peerTick outgoing peer+ (events, updated) <- peerTick outgoing peer -- Handle events liftIO $ mapM_ handleEvent events - gameLoop peer'+ gameLoop updated handleEvent :: PeerEvent -> IO () handleEvent = \case@@ -82,7 +83,7 @@ ```haskell -- Initiate connection (handshake happens automatically)-let peer' = peerConnect (peerIdFromAddr remoteAddr) now peer+let connecting = peerConnect (peerIdFromAddr remoteAddr) now peer -- The PeerConnected event fires when handshake completes ```@@ -130,16 +131,20 @@ ### Configuration +All config types have [optics](https://hackage.haskell.org/package/optics) labels via `OverloadedLabels`:+ ```haskell+{-# LANGUAGE OverloadedLabels #-}+import Optics ((&), (.~), (?~))+ let config = defaultNetworkConfig- { ncMaxClients = 32- , ncConnectionTimeoutMs = 10000.0- , ncKeepaliveIntervalMs = 1000.0- , ncMtu = 1200- , ncEnableConnectionMigration = True- , ncChannelConfigs = [unreliableChannel, reliableChannel]- , ncEncryptionKey = Just (EncryptionKey myKey) -- optional AEAD encryption- }+ & #ncMaxClients .~ 32+ & #ncConnectionTimeoutMs .~ 10000.0+ & #ncKeepaliveIntervalMs .~ 1000.0+ & #ncMtu .~ 1200+ & #ncEnableConnectionMigration .~ True+ & #ncChannelConfigs .~ [unreliableChannel, reliableChannel]+ & #ncEncryptionKey ?~ EncryptionKey myKey -- optional AEAD encryption ``` ---@@ -238,11 +243,10 @@ -- 10% packet loss simulateLoss 0.1 --- Packet duplication and out-of-order delivery+-- Packet duplication and out-of-order delivery (optics) let testCfg = defaultTestNetConfig- { tncDuplicateChance = 0.05 -- 5% chance of duplicating packets- , tncOutOfOrderChance = 0.1 -- 10% chance of reordering- }+ & #tncDuplicateChance .~ 0.05 -- 5% chance of duplicating packets+ & #tncOutOfOrderChance .~ 0.1 -- 10% chance of reordering ``` ---@@ -373,7 +377,7 @@ let acc = register npcId 2.0 $ register playerId 10.0 newPriorityAccumulator-let (selected, acc') = drainTop 1200 entitySize acc+let (selected, drained) = drainTop 1200 entitySize acc ``` ### Snapshot Interpolation@@ -383,8 +387,8 @@ ```haskell import GBNet.Replication.Interpolation -let buffer' = pushSnapshot serverTime state buffer-case sampleSnapshot renderTime buffer' of+let updated = pushSnapshot serverTime state buffer+case sampleSnapshot renderTime updated of Nothing -> waitForMoreSnapshots Just interpolated -> render interpolated ```@@ -417,9 +421,11 @@ Applications can query congestion pressure and adapt: ```haskell+import Optics (view)+ case peerStats peerId peer of Nothing -> pure () -- Peer not connected- Just stats -> case nsCongestionLevel stats of+ Just stats -> case view #nsCongestionLevel stats of CongestionNone -> sendFreely CongestionElevated -> reduceNonEssential CongestionHigh -> dropLowPriority
gbnet-hs.cabal view
@@ -1,6 +1,6 @@ cabal-version: 3.0 name: gbnet-hs-version: 0.2.1.0+version: 0.2.2.0 synopsis: Transport-level networking library with zero-copy Storable serialization description: A transport-level networking library providing reliable UDP with
src/GBNet/Channel.hs view
@@ -237,7 +237,7 @@ channelSend payload now ch | BS.length payload > ccMaxMessageSize (chConfig ch) = Left ChannelMessageTooLarge | bufferFull && ccBlockOnFull (chConfig ch) = Left ChannelBufferFull- | otherwise = Right (seqNum, ch')+ | otherwise = Right (seqNum, queued) where bufferFull = Map.size (chSendBuffer ch) >= ccMessageBufferSize (chConfig ch) seqNum = chLocalSequence ch@@ -255,7 +255,7 @@ if bufferFull then Map.deleteMin (chSendBuffer ch) else chSendBuffer ch- ch' =+ queued = ch & #chLocalSequence .~ (seqNum + 1)@@ -278,13 +278,11 @@ if cmReliable msg then -- Reliable: keep in buffer for retransmit- let msg' = msg & #cmRetryCount .~ 1- ch' = ch & #chSendBuffer %~ Map.insert seqNum msg'- in Just (msg, ch')+ let marked = msg & #cmRetryCount .~ 1+ in Just (msg, ch & #chSendBuffer %~ Map.insert seqNum marked) else -- Unreliable: remove from buffer immediately (fire and forget)- let ch' = ch & #chSendBuffer %~ Map.delete seqNum- in Just (msg, ch')+ Just (msg, ch & #chSendBuffer %~ Map.delete seqNum) | otherwise -> Nothing -- Already sent, waiting for ack or retransmit -- | Get messages that need retransmission based on RTO.@@ -309,14 +307,14 @@ ) | elapsedMs (cmSendTime msg) now >= rtoMs = -- Needs retransmit- let msg' = msg & #cmSendTime .~ now & #cmRetryCount %~ (+ 1)- c' =+ let retried = msg & #cmSendTime .~ now & #cmRetryCount %~ (+ 1)+ in ( msg : acc, c & #chSendBuffer- %~ Map.insert seqNum msg'+ %~ Map.insert seqNum retried & #chTotalRetransmits %~ (+ 1)- in (msg : acc, c')+ ) | otherwise = (acc, c) -- | Process a received message. Returns updated channel.@@ -349,10 +347,10 @@ & #chTotalReceived %~ (+ 1) ReliableOrdered ->- let ch' = ch & #chPendingAck %~ (seqNum :)- in if seqNum == chOrderedExpected ch'- then deliverOrdered payload ch'- else bufferOrdered seqNum payload now ch'+ let withAck = ch & #chPendingAck %~ (seqNum :)+ in if seqNum == chOrderedExpected withAck+ then deliverOrdered payload withAck+ else bufferOrdered seqNum payload now withAck ReliableSequenced -> if sequenceGreaterThan seqNum (chRemoteSequence ch) then@@ -375,7 +373,7 @@ -- | Deliver message and flush any buffered consecutive messages. deliverOrdered :: BS.ByteString -> Channel -> Channel deliverOrdered payload ch =- let ch' =+ let delivered = ch & #chReceiveBuffer %~ (|> payload)@@ -383,7 +381,7 @@ %~ (+ 1) & #chTotalReceived %~ (+ 1)- in flushOrderedBuffer ch'+ in flushOrderedBuffer delivered -- | Buffer an out-of-order message for later delivery. bufferOrdered :: SequenceNum -> BS.ByteString -> MonoTime -> Channel -> Channel@@ -399,7 +397,7 @@ case Map.lookup (chOrderedExpected ch) (chOrderedReceiveBuffer ch) of Nothing -> ch Just (payload, _) ->- let ch' =+ let flushed = ch & #chOrderedReceiveBuffer %~ Map.delete (chOrderedExpected ch)@@ -409,7 +407,7 @@ %~ (+ 1) & #chTotalReceived %~ (+ 1)- in flushOrderedBuffer ch'+ in flushOrderedBuffer flushed -- | Acknowledge a message by sequence number. acknowledgeMessage :: SequenceNum -> Channel -> Channel
src/GBNet/Congestion.hs view
@@ -251,8 +251,8 @@ let maxRate = ccBaseSendRate cc * maxSendRateMultiplier newRate = min maxRate (ccCurrentSendRate cc + sendRateIncrease) -- Halve recovery time after sustained good conditions- cc' = cc & #ccCurrentSendRate .~ newRate- in case ccLastGoodEntry cc' of+ ramped = cc & #ccCurrentSendRate .~ newRate+ in case ccLastGoodEntry ramped of Just goodEntry -> let elapsed = elapsedMs goodEntry now / 1000.0 intervals = floor (elapsed / recoveryHalveIntervalSecs) :: Int@@ -261,14 +261,14 @@ let newRecovery = max minRecoverySecs- (ccAdaptiveRecoverySecs cc' / (2.0 ^ intervals))- in cc'+ (ccAdaptiveRecoverySecs ramped / (2.0 ^ intervals))+ in ramped & #ccAdaptiveRecoverySecs .~ newRecovery & #ccLastGoodEntry ?~ now- else cc'- Nothing -> cc'+ else ramped+ Nothing -> ramped CongestionBad | not isBad -> case ccGoodConditionsStart cc of@@ -360,21 +360,21 @@ -- | Called when bytes are acknowledged. cwOnAck :: Int -> CongestionWindow -> CongestionWindow cwOnAck bytes cw =- let cw' = cw & #cwBytesInFlight %~ (\b -> b - fromIntegral (min bytes (fromIntegral b)))- in case cwPhase cw' of+ let deflated = cw & #cwBytesInFlight %~ (\b -> b - fromIntegral (min bytes (fromIntegral b)))+ in case cwPhase deflated of SlowStart ->- let newCwnd = cwCwnd cw' + fromIntegral bytes- in if newCwnd >= cwSsthresh cw'- then cw' & #cwCwnd .~ newCwnd & #cwPhase .~ Avoidance- else cw' & #cwCwnd .~ newCwnd+ let newCwnd = cwCwnd deflated + fromIntegral bytes+ in if newCwnd >= cwSsthresh deflated+ then deflated & #cwCwnd .~ newCwnd & #cwPhase .~ Avoidance+ else deflated & #cwCwnd .~ newCwnd Avoidance- | cwCwnd cw' > 0 ->+ | cwCwnd deflated > 0 -> -- Additive increase: cwnd += mtu * bytes / cwnd- let increase = fromIntegral (cwMtu cw') * fromIntegral bytes / cwCwnd cw'- in cw' & #cwCwnd %~ (+ increase)- | otherwise -> cw'+ let increase = fromIntegral (cwMtu deflated) * fromIntegral bytes / cwCwnd deflated+ in deflated & #cwCwnd %~ (+ increase)+ | otherwise -> deflated Recovery ->- cw' -- Conservative in recovery+ deflated -- Conservative in recovery -- | Called on packet loss detection. cwOnLoss :: CongestionWindow -> CongestionWindow@@ -470,8 +470,8 @@ -- | Record bytes at the given time. btRecord :: Int -> MonoTime -> BandwidthTracker -> BandwidthTracker btRecord bytes now bt =- let bt' = bt & #btWindow %~ (Seq.|> (now, bytes)) & #btTotalBytes %~ (+ bytes)- in btCleanup now bt'+ let recorded = bt & #btWindow %~ (Seq.|> (now, bytes)) & #btTotalBytes %~ (+ bytes)+ in btCleanup now recorded -- | Get bytes per second. btBytesPerSecond :: BandwidthTracker -> Double
src/GBNet/Connection.hs view
@@ -396,8 +396,8 @@ Just channel -> case Channel.channelSend payload now channel of Left chErr -> Left (ErrChannelError chErr)- Right (_msgSeq, channel') ->- Right $ modifyChannel idx (const channel') conn+ Right (_msgSeq, queued) ->+ Right $ modifyChannel idx (const queued) conn where idx = channelIdToInt channelId @@ -407,8 +407,8 @@ case IntMap.lookup idx (connChannels conn) of Nothing -> ([], conn) Just channel ->- let (msgs, channel') = Channel.channelReceive channel- in (msgs, modifyChannel idx (const channel') conn)+ let (msgs, drained) = Channel.channelReceive channel+ in (msgs, modifyChannel idx (const drained) conn) where idx = channelIdToInt channelId @@ -469,20 +469,20 @@ .~ True -- Process ACKs (extend 32-bit wire format to 64-bit) ackBits64 = fromIntegral (ackBitfield header) :: Word64- (ackResult, rel') = Rel.processAcks (ack header) ackBits64 now (connReliability conn1)+ (ackResult, updatedRel) = Rel.processAcks (ack header) ackBits64 now (connReliability conn1) ackedPairs = Rel.arAcked ackResult fastRetransmits = Rel.arFastRetransmit ackResult- conn2 = conn1 & #connReliability .~ rel'+ conn2 = conn1 & #connReliability .~ updatedRel -- Feed ack/loss info to cwnd conn3 = case connCwnd conn2 of Just cwVal -> let hasLoss = not (null fastRetransmits) ackedBytes = length ackedPairs * ncMtu (connConfig conn2)- cwVal'+ updatedCw | hasLoss = cwOnLoss cwVal | ackedBytes > 0 = cwOnAck ackedBytes cwVal | otherwise = cwVal- in conn2 & #connCwnd ?~ cwVal'+ in conn2 & #connCwnd ?~ updatedCw Nothing -> conn2 -- Acknowledge messages on channels conn4 = foldl' (\c (chId, chSeq) -> modifyChannel (channelIdToInt chId) (Channel.acknowledgeMessage chSeq) c) conn3 ackedPairs@@ -532,8 +532,8 @@ updateConnectedPure now conn0 = let -- Update congestion cfg = connConfig conn0- cong' = ccRefillBudget (ncMtu cfg) $ ccUpdate (nsPacketLoss (connStats conn0)) (nsRtt (connStats conn0)) now (connCongestion conn0)- conn1 = conn0 & #connCongestion .~ cong'+ congestion = ccRefillBudget (ncMtu cfg) $ ccUpdate (nsPacketLoss (connStats conn0)) (nsRtt (connStats conn0)) now (connCongestion conn0)+ conn1 = conn0 & #connCongestion .~ congestion -- Update cwnd pacing and check for slow start restart conn2 = case connCwnd conn1 of Just cwVal ->@@ -560,36 +560,34 @@ windowLevel = maybe CongestionNone cwCongestionLevel (connCwnd conn6) congLevel = max binaryLevel windowLevel -- Update stats- rel' = connReliability conn6- conn7 =- conn6- & #connStats- % #nsRtt- .~ Rel.srttMs rel'- & #connStats- % #nsPacketLoss- .~ Rel.packetLossPercent rel'- & #connStats- % #nsBandwidthUp- .~ btBytesPerSecond (connBandwidthUp conn6)- & #connStats- % #nsBandwidthDown- .~ btBytesPerSecond (connBandwidthDown conn6)- & #connStats- % #nsConnectionQuality- .~ assessConnectionQuality (Rel.srttMs rel') (Rel.packetLossPercent rel' * 100)- & #connStats- % #nsCongestionLevel- .~ congLevel- & #connPendingAck- .~ False- in conn7+ reliability = connReliability conn6+ in conn6+ & #connStats+ % #nsRtt+ .~ Rel.srttMs reliability+ & #connStats+ % #nsPacketLoss+ .~ Rel.packetLossPercent reliability+ & #connStats+ % #nsBandwidthUp+ .~ btBytesPerSecond (connBandwidthUp conn6)+ & #connStats+ % #nsBandwidthDown+ .~ btBytesPerSecond (connBandwidthDown conn6)+ & #connStats+ % #nsConnectionQuality+ .~ assessConnectionQuality (Rel.srttMs reliability) (Rel.packetLossPercent reliability * 100)+ & #connStats+ % #nsCongestionLevel+ .~ congLevel+ & #connPendingAck+ .~ False -- | Process outgoing messages from channels. processChannelOutput :: MonoTime -> Connection -> Connection processChannelOutput now conn =- let conn' = conn & #connDataSentThisTick .~ False- in foldl' (processChannelIdx now) conn' (connChannelPriority conn')+ let reset = conn & #connDataSentThisTick .~ False+ in foldl' (processChannelIdx now) reset (connChannelPriority reset) processChannelIdx :: MonoTime -> Connection -> Int -> Connection processChannelIdx now conn chIdx =@@ -602,10 +600,10 @@ | not canSendBinary = conn | otherwise = case tryDequeue of Nothing -> conn- Just (msg, channel', wireData, wireSize, isReliable)+ Just (msg, dequeued, wireData, wireSize, isReliable) | not (cwndAllows isReliable wireSize) -> conn | otherwise ->- processChannelMessages now (emitPacket conn msg channel' wireData wireSize isReliable) chIdx+ processChannelMessages now (emitPacket conn msg dequeued wireData wireSize isReliable) chIdx where mtu = ncMtu (connConfig conn) canSendBinary = ccCanSend 0 mtu (connCongestion conn)@@ -617,16 +615,16 @@ tryDequeue = do channel <- IntMap.lookup chIdx (connChannels conn)- (msg, channel') <- Channel.getOutgoingMessage channel+ (msg, dequeued) <- Channel.getOutgoingMessage channel let msgSeqRaw = unSequenceNum (cmSequence msg) -- Payload header: channel (3 bits) | is_fragment (1 bit), then 16-bit channel seq headerByte = fromIntegral chIdx .&. 0x07 seqHi = fromIntegral (msgSeqRaw `shiftR` 8) :: Word8 seqLo = fromIntegral (msgSeqRaw .&. 0xFF) :: Word8 wireData = BS.cons headerByte $ BS.cons seqHi $ BS.cons seqLo (cmData msg)- pure (msg, channel', wireData, BS.length wireData, Channel.channelIsReliable channel)+ pure (msg, dequeued, wireData, BS.length wireData, Channel.channelIsReliable channel) - emitPacket c msg channel' wireData wireSize isReliable =+ emitPacket c msg dequeued wireData wireSize isReliable = let header = createHeaderInternal c pkt = OutgoingPacket@@ -634,9 +632,9 @@ opType = Payload, opPayload = wireData }- cong' = ccDeductBudget wireSize (connCongestion c)- cwnd' = fmap (cwOnSend wireSize now) (connCwnd c)- rel'+ congestion = ccDeductBudget wireSize (connCongestion c)+ cwnd = fmap (cwOnSend wireSize now) (connCwnd c)+ reliability | isReliable = Rel.onPacketSent (connLocalSeq c)@@ -648,17 +646,17 @@ | otherwise = connReliability c in c & #connChannels- %~ IntMap.insert chIdx channel'+ %~ IntMap.insert chIdx dequeued & #connSendQueue %~ (Seq.|> pkt) & #connLocalSeq %~ (+ 1) & #connCongestion- .~ cong'+ .~ congestion & #connCwnd- .~ cwnd'+ .~ cwnd & #connReliability- .~ rel'+ .~ reliability & #connDataSentThisTick .~ True
src/GBNet/Fragment.hs view
@@ -202,13 +202,13 @@ | idx >= fbFragmentCount buf = (False, buf) | Map.member idx (fbFragments buf) = (isComplete buf, buf) -- Already have this fragment | otherwise =- let buf' =+ let inserted = buf & #fbFragments %~ Map.insert idx dat & #fbTotalSize %~ (+ BS.length dat)- in (isComplete buf', buf')+ in (isComplete inserted, inserted) -- | Check if all fragments received. isComplete :: FragmentBuffer -> Bool@@ -272,22 +272,22 @@ Just b -> b Nothing -> newFragmentBuffer (fhFragmentCount header) now -- Insert fragment- (complete, buf') = insertFragment (fhFragmentIndex header) fragData buf+ (complete, inserted) = insertFragment (fhFragmentIndex header) fragData buf asm3 = asm2 & #faBuffers- %~ Map.insert msgId buf'+ %~ Map.insert msgId inserted & #faCurrentBufferSize %~ (+ fragSize) in if complete then- let result = assembleFragments buf'+ let result = assembleFragments inserted asm4 = asm3 & #faBuffers %~ Map.delete msgId & #faCurrentBufferSize- %~ subtract (fbTotalSize buf')+ %~ subtract (fbTotalSize inserted) in (result, asm4) else (Nothing, asm3) @@ -378,7 +378,7 @@ (Nothing, md) _ -> let probe = (mdMinMtu md + mdMaxMtu md) `div` 2- md' =+ probing = md & #mdCurrentProbe .~ probe@@ -386,7 +386,7 @@ ?~ now & #mdAttempts %~ (+ 1)- in (Just probe, md')+ in (Just probe, probing) -- | Called when probe succeeded (ack received). onProbeSuccess :: Int -> MtuDiscovery -> MtuDiscovery
src/GBNet/Net.hs view
@@ -156,8 +156,8 @@ result <- liftIO $ socketSendTo bytes toAddr now (nsSocket st) case result of Left err -> pure $ Left (NetSendFailed (show err))- Right (_, sock') -> do- modifyNetState $ #nsSocket .~ sock'+ Right (_, updatedSock) -> do+ modifyNetState $ #nsSocket .~ updatedSock pure $ Right () netRecv = do
src/GBNet/Peer.hs view
@@ -164,8 +164,7 @@ | Map.member peerId (npConnections peer) = peer -- Already connected | Map.member peerId (npPending peer) = peer -- Already pending | otherwise =- -- Generate client salt- let (salt, rng') = nextRandom (npRngState peer)+ let (salt, rng) = nextRandom (npRngState peer) pending = PendingConnection { pcDirection = Outbound,@@ -175,14 +174,16 @@ pcRetryCount = 0, pcLastRetry = now }- peer' =- peer- & #npPending- %~ Map.insert peerId pending- & #npRngState- .~ rng'- in -- Queue connection request- queueControlPacket ConnectionRequest BS.empty peerId peer'+ in queueControlPacket+ ConnectionRequest+ BS.empty+ peerId+ ( peer+ & #npPending+ %~ Map.insert peerId pending+ & #npRngState+ .~ rng+ ) -- | Disconnect a specific peer (pure). -- Transitions the connection to Disconnecting state for graceful shutdown@@ -227,16 +228,19 @@ -- Post-handshake packets (Payload, Keepalive, Disconnect) are encrypted -- when the connection has an encryption key configured. drainAllConnectionQueues :: MonoTime -> NetPeer -> NetPeer-drainAllConnectionQueues _now peer =+drainAllConnectionQueues now peer = Map.foldlWithKey' drainOne (peer & #npConnections .~ Map.empty) (npConnections peer) where protocolId = ncProtocolId (npConfig peer) - drainOne p peerId conn =- let (connPackets, conn') = Conn.drainSendQueue conn- (rawPackets, conn'') = encryptOutgoing peerId protocolId conn' connPackets- p' = p & #npConnections %~ Map.insert peerId conn''- in foldl' (flip queueRawPacket) p' rawPackets+ drainOne acc peerId conn =+ let (connPackets, drained) = Conn.drainSendQueue conn+ (rawPackets, encrypted) = encryptOutgoing peerId protocolId drained connPackets+ !bytesSent = sum (map (BS.length . rpData) rawPackets)+ in foldl'+ (flip queueRawPacket)+ (acc & #npConnections %~ Map.insert peerId (Conn.recordBytesSent bytesSent now encrypted))+ rawPackets -- | Encrypt outgoing packets and update connection nonce state. -- Handshake packets remain plaintext; post-handshake packets are encrypted@@ -247,13 +251,15 @@ Conn.Connection -> [OutgoingPacket] -> ([RawPacket], Conn.Connection)-encryptOutgoing peerId protocolId conn0 =- foldl' encryptOne ([], conn0)+encryptOutgoing peerId protocolId conn0 packets =+ let (revPackets, finalConn) = foldl' encryptOne ([], conn0) packets+ in (reverse revPackets, finalConn) where- encryptOne (!acc, !conn) (OutgoingPacket hdr ptype payload) =+ encryptOne (!revAcc, !conn) (OutgoingPacket hdr ptype payload) = let header = hdr {packetType = ptype} pkt = Packet {pktHeader = header, pktPayload = payload} serialized = serializePacket pkt+ raw plainBytes = RawPacket peerId (appendCrc32 plainBytes) in case (connEncryptionKey conn, isPostHandshake ptype) of (Just key, True) -> let nonce = connSendNonce conn@@ -261,20 +267,13 @@ payloadBytes = BS.drop packetHeaderByteSize serialized in case encrypt key nonce protocolId payloadBytes of Left _ ->- -- Encryption failed; send plaintext as fallback- let raw = appendCrc32 serialized- in (acc ++ [RawPacket peerId raw], conn)+ (raw serialized : revAcc, conn) Right encrypted ->- let raw = appendCrc32 (headerBytes <> encrypted)- conn' =- conn- & #connSendNonce- .~ NonceCounter (unNonceCounter nonce + 1)- in (acc ++ [RawPacket peerId raw], conn')+ ( raw (headerBytes <> encrypted) : revAcc,+ conn & #connSendNonce .~ NonceCounter (unNonceCounter nonce + 1)+ ) _ ->- -- Plaintext: no key or handshake packet- let raw = appendCrc32 serialized- in (acc ++ [RawPacket peerId raw], conn)+ (raw serialized : revAcc, conn) -- | Whether a packet type is post-handshake (should be encrypted). isPostHandshake :: PacketType -> Bool@@ -312,9 +311,8 @@ peerShutdownM peer = do now <- getMonoTime let peerIds = Map.keys (npConnections peer)- -- Disconnect each connection through the proper state machine- peer' = foldr (\pid p -> withConnection pid (Conn.disconnect ReasonRequested now) p) peer peerIds- (outgoing, _) = drainPeerSendQueue (drainAllConnectionQueues now peer')+ disconnected = foldr (\pid p -> withConnection pid (Conn.disconnect ReasonRequested now) p) peer peerIds+ (outgoing, _) = drainPeerSendQueue (drainAllConnectionQueues now disconnected) peerSendAllM outgoing netClose @@ -361,11 +359,13 @@ MonoTime -> NetPeer -> ([PeerEvent], NetPeer)-processPacketsPure packets now peer = foldl' go ([], peer) packets+processPacketsPure packets now peer =+ let (revEvents, finalPeer) = foldl' go ([], peer) packets+ in (reverse revEvents, finalPeer) where- go (evts, p) (pid, dat) =- let (evts', p') = handlePacket pid dat now p- in (evts ++ evts', p')+ go (revEvts, p) (pid, dat) =+ let (newEvts, updated) = handlePacket pid dat now p+ in (prependReversed newEvts revEvts, updated) -- | Handle a single received packet (pure). -- For post-handshake packets from connections with encryption keys,@@ -381,39 +381,40 @@ case parsePacket dat of Nothing -> ([], peer) Just pkt ->- let ptype = packetType (pktHeader pkt)+ let !bytesReceived = BS.length dat+ ptype = packetType (pktHeader pkt)+ recordBytes = Conn.recordBytesReceived bytesReceived now in case (isPostHandshake ptype, lookupConnectionKey peerId peer) of (True, Just (key, conn)) ->- let protocolId = ncProtocolId (npConfig peer)- encPayload = pktPayload pkt- in case decrypt key protocolId encPayload of- Left _ ->- -- Decryption failed: increment stat, drop packet- let conn' =- conn- & #connStats- % #nsDecryptionFailures- %~ (+ 1)- peer' = peer & #npConnections %~ Map.insert peerId conn'- in ([], peer')- Right (plaintext, NonceCounter recvNonce) ->- -- Anti-replay check- case connRecvNonceMax conn of- Just maxNonce- | recvNonce <= maxNonce ->- ([], peer) -- Nonce replay, drop- _ ->- let conn' =- conn- & #connRecvNonceMax- ?~ recvNonce- peer' = peer & #npConnections %~ Map.insert peerId conn'- decryptedPkt = pkt {pktPayload = plaintext}- in handlePacketByType peerId decryptedPkt now ptype peer'+ dispatchEncrypted pkt ptype key (recordBytes conn) _ ->- -- No encryption or handshake packet: process as plaintext- handlePacketByType peerId pkt now ptype peer+ handlePacketByType+ peerId+ pkt+ now+ ptype+ (withConnection peerId recordBytes peer)+ where+ putConn c = peer & #npConnections %~ Map.insert peerId c + dispatchEncrypted pkt ptype key conn =+ let protocolId = ncProtocolId (npConfig peer)+ in case decrypt key protocolId (pktPayload pkt) of+ Left _ ->+ ([], putConn (conn & #connStats % #nsDecryptionFailures %~ (+ 1)))+ Right (plaintext, NonceCounter recvNonce) ->+ case connRecvNonceMax conn of+ Just maxNonce+ | recvNonce <= maxNonce -> ([], putConn conn)+ _ ->+ let decryptedPkt = pkt {pktPayload = plaintext}+ in handlePacketByType+ peerId+ decryptedPkt+ now+ ptype+ (putConn (conn & #connRecvNonceMax ?~ recvNonce))+ -- | Look up a connection's encryption key and connection for a peer. lookupConnectionKey :: PeerId ->@@ -448,20 +449,19 @@ ConnectionAccepted -> handleConnectionAccepted peerId now peer ConnectionDenied -> let reason = decodeDenyReason (pktPayload pkt)- peer' = removePending peerId peer- in ([PeerDisconnected peerId (denyToDisconnectReason reason)], peer')+ in ([PeerDisconnected peerId (denyToDisconnectReason reason)], removePending peerId peer) Disconnect -> handleDisconnect peerId peer Payload -> if Map.member peerId (npConnections peer) then handlePayload peerId pkt now peer else handleMigration peerId pkt now peer Keepalive ->- let peer' =- withConnection- peerId- (Conn.touchRecvTime now . processIncomingHeader (pktHeader pkt) now)- peer- in ([], peer')+ ( [],+ withConnection+ peerId+ (Conn.touchRecvTime now . processIncomingHeader (pktHeader pkt) now)+ peer+ ) -- | Handle payload packet (pure). -- Routes messages through the channel system for proper ordering/dedup.@@ -470,25 +470,23 @@ case Map.lookup peerId (npConnections peer) of Nothing -> ([], peer) Just conn ->- let conn' = Conn.touchRecvTime now $ processIncomingHeader (pktHeader pkt) now conn+ let processed = Conn.touchRecvTime now $ processIncomingHeader (pktHeader pkt) now conn payload = pktPayload pkt+ putConn c = peer & #npConnections %~ Map.insert peerId c in case BS.uncons payload of- Nothing ->- ([], peer & #npConnections %~ Map.insert peerId conn')+ Nothing -> ([], putConn processed) Just (headerByte, rest) -> let (channel, isFragment) = decodePayloadHeader headerByte in if isFragment- then- let peer' = peer & #npConnections %~ Map.insert peerId conn'- in handleFragment peerId channel rest now peer'+ then handleFragment peerId channel rest now (putConn processed) else let finalConn- | BS.length payload < minPayloadSize = conn'+ | BS.length payload < minPayloadSize = processed | otherwise = case Proto.decodeChannelSeq rest of- Nothing -> conn'+ Nothing -> processed Just (chSeq, msgData) ->- receiveIncomingPayload channel chSeq msgData now conn'- in ([], peer & #npConnections %~ Map.insert peerId finalConn)+ receiveIncomingPayload channel chSeq msgData now processed+ in ([], putConn finalConn) -- | Handle a fragment, reassembling if complete (pure). -- After reassembly, routes through the channel system for ordering/dedup.@@ -500,15 +498,15 @@ (newFragmentAssembler fragmentTimeoutMs fragmentMaxBufferSize) peerId assemblers- (maybeComplete, assembler') = processFragment fragData now assembler- peer' = peer & #npFragmentAssemblers .~ Map.insert peerId assembler' assemblers+ (maybeComplete, updated) = processFragment fragData now assembler+ withAssembler = peer & #npFragmentAssemblers .~ Map.insert peerId updated assemblers in case maybeComplete of- Nothing -> ([], peer')+ Nothing -> ([], withAssembler) Just completeData -> case Proto.decodeChannelSeq completeData of- Nothing -> ([], peer')+ Nothing -> ([], withAssembler) Just (chSeq, msgData) ->- ([], withConnection peerId (receiveIncomingPayload channel chSeq msgData now) peer')+ ([], withConnection peerId (receiveIncomingPayload channel chSeq msgData now) withAssembler) -- | Try to migrate an existing connection to a new address, or ignore the packet (pure). handleMigration :: PeerId -> Packet -> MonoTime -> NetPeer -> ([PeerEvent], NetPeer)@@ -523,7 +521,7 @@ | elapsedMs lastMigration now < migrationCooldownMs -> ([], peer) -- Still in cooldown _ ->- let peer' =+ let migrated = peer & #npConnections %~ (Map.insert newPeerId conn . Map.delete oldPeerId)@@ -535,31 +533,30 @@ Just asm -> Map.insert newPeerId asm $ Map.delete oldPeerId fa ) event = PeerMigrated oldPeerId newPeerId- (payloadEvents, peer'') = handlePayload newPeerId pkt now peer'- in (event : payloadEvents, peer'')+ (payloadEvents, withPayload) = handlePayload newPeerId pkt now migrated+ in (event : payloadEvents, withPayload) -- | Update all connections and collect messages/disconnects (pure). -- Uses reverse accumulator to avoid O(n^2) list appending. updateConnections :: MonoTime -> NetPeer -> ([PeerEvent], NetPeer) updateConnections now peer = let conns = npConnections peer- (revEvents, conns', disconnectedIds) = Map.foldlWithKey' updateOne ([], Map.empty, []) conns- peer' = foldl' (flip cleanupPeer) (peer & #npConnections .~ conns') disconnectedIds- in (reverse revEvents, peer')+ (revEvents, updatedConns, disconnectedIds) = Map.foldlWithKey' updateOne ([], Map.empty, []) conns+ cleaned = foldl' (flip cleanupPeer) (peer & #npConnections .~ updatedConns) disconnectedIds+ -- Sweep stale migration cooldown entries to prevent unbounded growth+ sweptCooldowns = Map.filter (\t -> elapsedMs t now < migrationCooldownMs) (npMigrationCooldowns cleaned)+ in (reverse revEvents, cleaned & #npMigrationCooldowns .~ sweptCooldowns) where updateOne (revEvts, connsAcc, discs) peerId conn = case Conn.updateTick now conn of Left _err ->- -- Connection timed out (PeerDisconnected peerId ReasonTimeout : revEvts, connsAcc, peerId : discs)- Right conn'- | Conn.connectionState conn' == Conn.Disconnected ->- -- Graceful disconnect complete+ Right updated+ | Conn.connectionState updated == Conn.Disconnected -> (PeerDisconnected peerId ReasonRequested : revEvts, connsAcc, peerId : discs) | otherwise ->- -- Collect messages from all channels- let (msgs, conn'') = collectMessages peerId conn'- in (prependReversed msgs revEvts, Map.insert peerId conn'' connsAcc, discs)+ let (msgs, withMsgs) = collectMessages peerId updated+ in (prependReversed msgs revEvts, Map.insert peerId withMsgs connsAcc, discs) collectMessages peerId conn = let numChannels = Conn.channelCount conn@@ -569,9 +566,9 @@ | ch >= maxCh = (reverse revAcc, conn) | otherwise = let chId = ChannelId ch- (msgs, conn') = Conn.receiveMessage chId conn+ (msgs, received) = Conn.receiveMessage chId conn evts = map (PeerMessage peerId chId) msgs- in collectFromChannels peerId (ch + 1) maxCh conn' (prependReversed evts revAcc)+ in collectFromChannels peerId (ch + 1) maxCh received (prependReversed evts revAcc) -- | Prepend a list in reverse onto an accumulator. O(length xs). -- Used for efficient reverse-accumulator pattern.@@ -584,15 +581,20 @@ let outbound = Map.toList $ Map.filter (\p -> pcDirection p == Outbound) (npPending peer) in foldl' (retryOne now) peer outbound where- retryOne t p (peerId, pending) =+ retryOne t acc (peerId, pending) = let elapsed = elapsedMs (pcLastRetry pending) t- retryInterval = ncConnectionRequestTimeoutMs (npConfig p) / fromIntegral (ncConnectionRequestMaxRetries (npConfig p) + 1)- in if elapsed > retryInterval && pcRetryCount pending < ncConnectionRequestMaxRetries (npConfig p)+ retryInterval = ncConnectionRequestTimeoutMs (npConfig acc) / fromIntegral (ncConnectionRequestMaxRetries (npConfig acc) + 1)+ in if elapsed > retryInterval && pcRetryCount pending < ncConnectionRequestMaxRetries (npConfig acc) then- let pending' = pending & #pcRetryCount %~ (+ 1) & #pcLastRetry .~ t- p' = p & #npPending %~ Map.insert peerId pending'- in queueControlPacket ConnectionRequest BS.empty peerId p'- else p+ queueControlPacket+ ConnectionRequest+ BS.empty+ peerId+ ( acc+ & #npPending+ %~ Map.insert peerId (pending & #pcRetryCount %~ (+ 1) & #pcLastRetry .~ t)+ )+ else acc -- | Cleanup expired pending connections (pure). cleanupPending :: MonoTime -> NetPeer -> ([PeerEvent], NetPeer)@@ -616,8 +618,8 @@ Just conn -> case Conn.sendMessage channel dat now conn of Left err -> Left err- Right conn' ->- Right (peer & #npConnections %~ Map.insert peerId conn')+ Right sent ->+ Right (peer & #npConnections %~ Map.insert peerId sent) -- | Broadcast a message to all connected peers. -- This queues the message and drains connection queues so packets are ready to send.@@ -633,9 +635,8 @@ -- Queue message to each connection's channel -- Best-effort: per-peer send failures (channel full, disconnected) are -- intentionally ignored so one failing peer doesn't block the broadcast.- peer' = foldl' (\p pid -> fromRight p (peerSend pid channel dat now p)) peer peerIds- in -- Drain connection queues to npSendQueue so packets are ready- drainAllConnectionQueues now peer'+ queued = foldl' (\p pid -> fromRight p (peerSend pid channel dat now p)) peer peerIds+ in drainAllConnectionQueues now queued -- | Get number of connected peers. peerCount :: NetPeer -> Int
src/GBNet/Peer/Handshake.hs view
@@ -58,8 +58,8 @@ handleNewConnectionRequest :: PeerId -> MonoTime -> NetPeer -> ([PeerEvent], NetPeer) handleNewConnectionRequest peerId now peer = let addrKey = sockAddrToKey (unPeerId peerId)- (allowed, rl') = rateLimiterAllow addrKey now (npRateLimiter peer)- peer1 = peer & #npRateLimiter .~ rl'+ (allowed, limiter) = rateLimiterAllow addrKey now (npRateLimiter peer)+ peer1 = peer & #npRateLimiter .~ limiter pendingSize = Map.size (npPending peer1) connSize = Map.size (npConnections peer1) maxClients = ncMaxClients (npConfig peer1)@@ -72,7 +72,7 @@ let reason = encodeDenyReason DenyServerFull in ([], queueControlPacket ConnectionDenied reason peerId peer1) | otherwise ->- let (salt, rng') = nextRandom (npRngState peer1)+ let (salt, rng) = nextRandom (npRngState peer1) newPend = PendingConnection { pcDirection = Inbound,@@ -87,7 +87,7 @@ & #npPending %~ Map.insert peerId newPend & #npRngState- .~ rng'+ .~ rng saltPayload = encodeSalt salt in ([], queueControlPacket ConnectionChallenge saltPayload peerId peer2) @@ -102,10 +102,12 @@ case decodeSalt (pktPayload pkt) of Nothing -> ([], peer) Just serverSalt ->- let p' = p & #pcServerSalt .~ serverSalt- peer' = peer & #npPending %~ Map.insert peerId p'- saltPayload = encodeSalt (pcClientSalt p')- in ([], queueControlPacket ConnectionResponse saltPayload peerId peer')+ let updated = p & #pcServerSalt .~ serverSalt+ saltPayload = encodeSalt (pcClientSalt updated)+ in ( [],+ queueControlPacket ConnectionResponse saltPayload peerId $+ peer & #npPending %~ Map.insert peerId updated+ ) -- | Handle connection response (we're inbound, received their response) (pure). handleConnectionResponse :: PeerId -> Packet -> MonoTime -> NetPeer -> ([PeerEvent], NetPeer)@@ -120,22 +122,22 @@ Just clientSalt | clientSalt == 0 || clientSalt == pcServerSalt p -> let reason = encodeDenyReason DenyInvalidChallenge- peer' =+ in ( [], queueControlPacket ConnectionDenied reason peerId $ removePending peerId peer- in ([], peer')+ ) | otherwise -> let conn = Conn.markConnected now $ Conn.touchRecvTime now $ newConnection (npConfig peer) clientSalt now- peer' =+ promoted = peer & #npConnections %~ Map.insert peerId conn & #npPending %~ Map.delete peerId- in ([PeerConnected peerId Inbound], queueControlPacket ConnectionAccepted BS.empty peerId peer')+ in ([PeerConnected peerId Inbound], queueControlPacket ConnectionAccepted BS.empty peerId promoted) -- | Handle connection accepted (we're outbound, they accepted) (pure). handleConnectionAccepted :: PeerId -> MonoTime -> NetPeer -> ([PeerEvent], NetPeer)@@ -149,19 +151,19 @@ Conn.markConnected now $ Conn.touchRecvTime now $ newConnection (npConfig peer) (pcClientSalt p) now- peer' =+ promoted = peer & #npConnections %~ Map.insert peerId conn & #npPending %~ Map.delete peerId- in ([PeerConnected peerId Outbound], peer')+ in ([PeerConnected peerId Outbound], promoted) -- | Handle disconnect packet (pure). handleDisconnect :: PeerId -> NetPeer -> ([PeerEvent], NetPeer) handleDisconnect peerId peer = if Map.member peerId (npConnections peer) then- let peer' = cleanupPeer peerId (peer & #npConnections %~ Map.delete peerId)- in ([PeerDisconnected peerId ReasonRequested], peer')+ let disconnected = cleanupPeer peerId (peer & #npConnections %~ Map.delete peerId)+ in ([PeerDisconnected peerId ReasonRequested], disconnected) else ([], removePending peerId peer)
src/GBNet/Peer/Internal.hs view
@@ -220,8 +220,8 @@ where go s 0 acc = (BS.pack (reverse acc), s) go s n acc =- let (r, s') = nextRandom s- in go s' (n - 1) (fromIntegral @Word64 @Word8 r : acc)+ let (r, advanced) = nextRandom s+ in go advanced (n - 1) (fromIntegral @Word64 @Word8 r : acc) -- ----------------------------------------------------------------------------- -- Helper functions for pure API@@ -255,7 +255,8 @@ raw = appendCrc32 (serializePacket pkt) in queueRawPacket (RawPacket pid raw) peer --- | Clean up per-peer state (fragment assemblers, migration cooldowns).+-- | Clean up per-peer state (fragment assemblers).+-- Migration cooldowns are swept separately in 'updateConnections'. cleanupPeer :: PeerId -> NetPeer -> NetPeer cleanupPeer peerId peer = peer & #npFragmentAssemblers %~ Map.delete peerId
src/GBNet/Reliability.hs view
@@ -241,25 +241,25 @@ newHighest | sequenceGreaterThan seqNum (sbSequence buf) = seqNum | otherwise = sbSequence buf- entries' = zeroCopyMutate (sbEntries buf) $ \mv ->+ mutated = zeroCopyMutate (sbEntries buf) $ \mv -> MV.unsafeWrite mv idx (SBEntry seqNum val)- in buf & #sbEntries .~ entries' & #sbSequence .~ newHighest+ in buf & #sbEntries .~ mutated & #sbSequence .~ newHighest {-# INLINE sbInsert #-} -- | Batched insert - O(n) with single thaw/freeze for n entries. sbInsertMany :: [(SequenceNum, a)] -> SequenceBuffer a -> SequenceBuffer a sbInsertMany [] buf = buf sbInsertMany items buf =- let (entries', newHighest) = zeroCopyMutate' (sbEntries buf) $ \mv ->+ let (mutated, newHighest) = zeroCopyMutate' (sbEntries buf) $ \mv -> let go _ !highest [] = return highest- go mv' !highest ((seqNum, val) : rest) = do+ go vec !highest ((seqNum, val) : rest) = do let !idx = seqToIndex seqNum buf- MV.unsafeWrite mv' idx (SBEntry seqNum val)- let !highest' =+ MV.unsafeWrite vec idx (SBEntry seqNum val)+ let !next = if sequenceGreaterThan seqNum highest then seqNum else highest- go mv' highest' rest+ go vec next rest in go mv (sbSequence buf) items- in buf & #sbEntries .~ entries' & #sbSequence .~ newHighest+ in buf & #sbEntries .~ mutated & #sbSequence .~ newHighest {-# INLINE sbInsertMany #-} -- | O(1) existence check via direct index.@@ -328,15 +328,15 @@ rbInsertMany :: [SequenceNum] -> ReceivedBuffer -> ReceivedBuffer rbInsertMany [] buf = buf rbInsertMany seqs buf =- let (v', newHighest) = zeroCopyMutateU' (rbSeqs buf) $ \mv ->+ let (mutated, newHighest) = zeroCopyMutateU' (rbSeqs buf) $ \mv -> go mv (rbHighest buf) seqs- in ReceivedBuffer v' newHighest+ in ReceivedBuffer mutated newHighest where go _ !highest [] = return highest go mv !highest (sn@(SequenceNum s) : rest) = do VUM.unsafeWrite mv (fromIntegral s .&. receivedBufferMask) s- let highest' = if sequenceGreaterThan sn highest then sn else highest- go mv highest' rest+ let next = if sequenceGreaterThan sn highest then sn else highest+ go mv next rest {-# INLINE rbInsertMany #-} -- SentPacketRecord@@ -419,13 +419,13 @@ spbInsert :: SequenceNum -> SentPacketRecord -> SentPacketBuffer -> SentPacketBuffer spbInsert seqNum record buf = let idx = seqToIdx seqNum- (entries', countDelta) = zeroCopyMutate' (spbEntries buf) $ \mvec -> do+ (mutated, countDelta) = zeroCopyMutate' (spbEntries buf) $ \mvec -> do old <- MV.unsafeRead mvec idx MV.unsafeWrite mvec idx (RingEntry seqNum record) return $! case old of RingEmpty -> 1 :: Int _ -> 0- in buf & #spbEntries .~ entries' & #spbCount %~ (+ countDelta)+ in buf & #spbEntries .~ mutated & #spbCount %~ (+ countDelta) {-# INLINE spbInsert #-} -- | O(1) delete by sequence number.@@ -434,9 +434,9 @@ case V.unsafeIndex (spbEntries buf) idx of RingEntry storedSeq _ | storedSeq == seqNum ->- let entries' = zeroCopyMutate (spbEntries buf) $ \mvec ->+ let cleared = zeroCopyMutate (spbEntries buf) $ \mvec -> MV.unsafeWrite mvec idx RingEmpty- in buf & #spbEntries .~ entries' & #spbCount %~ subtract 1+ in buf & #spbEntries .~ cleared & #spbCount %~ subtract 1 _ -> buf where idx = seqToIdx seqNum@@ -564,7 +564,7 @@ ReliableEndpoint -> ReliableEndpoint onPacketSent seqNum sendTime channelId channelSeq size ep =- let ep' =+ let withRoom = if spbCount (reSentPackets ep) >= reMaxInFlight ep then evictWorstInFlight ep else ep@@ -576,7 +576,7 @@ sprSize = size, sprNackCount = 0 }- in ep'+ in withRoom & #reSentPackets %~ spbInsert seqNum record & #reTotalSent@@ -635,22 +635,22 @@ (acked, retrans, mutations, rttSum, rttCount, bytesAcked) = foldl' (processOne buf) ([], [], [], 0.0, 0 :: Int, 0) [0 .. fromIntegral ackBitsWindow] -- Phase 2: Apply all mutations in one ST block- buf' = applyMutations mutations buf+ mutatedBuf = applyMutations mutations buf -- Update RTT with average of all samples- ep' = case rttCount of+ withRtt = case rttCount of 0 -> ep _ -> updateRtt (rttSum / fromIntegral rttCount) ep -- Record loss samples for each ack (success = not lost)- ep'' = foldl' (\e _ -> recordLossSample False e) ep' acked- ep''' =- ep''+ withLoss = foldl' (\e _ -> recordLossSample False e) withRtt acked+ in ( AckResult acked retrans,+ withLoss & #reSentPackets- .~ buf'+ .~ mutatedBuf & #reTotalAcked .~ (reTotalAcked ep + fromIntegral (length acked)) & #reBytesAcked .~ (reBytesAcked ep + fromIntegral bytesAcked)- in (AckResult acked retrans, ep''')+ ) where processOne buf (!acked, !retrans, !muts, !rttSum, !rttCnt, !bytes) i | i == 0 = checkSeq ackSeq True buf acked retrans muts rttSum rttCnt bytes@@ -671,19 +671,19 @@ | otherwise -> let idx = seqToIdx seqNum newNack = min 255 (sprNackCount record + 1)- retrans' =+ updatedRetrans = if newNack == fastRetransmitThreshold then (sprChannelId record, sprChannelSequence record) : retrans else retrans- in (acked, retrans', MutNack idx seqNum record newNack : muts, rttSum, rttCnt, bytes)+ in (acked, updatedRetrans, MutNack idx seqNum record newNack : muts, rttSum, rttCnt, bytes) -- | Apply all mutations in one ST block via zero-copy mutation. applyMutations :: [BufferMutation] -> SentPacketBuffer -> SentPacketBuffer applyMutations [] buf = buf applyMutations muts buf =- let (entries', deletions) = zeroCopyMutate' (spbEntries buf) $ \mvec ->+ let (mutated, deletions) = zeroCopyMutate' (spbEntries buf) $ \mvec -> applyAll mvec muts 0- in buf & #spbEntries .~ entries' & #spbCount %~ subtract deletions+ in buf & #spbEntries .~ mutated & #spbCount %~ subtract deletions where applyAll _ [] !dels = return dels applyAll mvec (MutDelete idx : rest) !dels = do
src/GBNet/Replication/Delta.hs view
@@ -146,7 +146,7 @@ a -> DeltaTracker a -> (BS.ByteString, DeltaTracker a)-deltaEncode seq' current tracker =+deltaEncode seqNum current tracker = let -- Encode based on whether we have a confirmed baseline encoded = case dtConfirmed tracker of Just (baseSeq, baseline) ->@@ -158,33 +158,33 @@ -- Store pending snapshot pending = dtPending tracker- pending' =+ updated = if Seq.length pending >= dtMaxPending tracker- then Seq.drop 1 pending Seq.|> (seq', current)- else pending Seq.|> (seq', current)-- tracker' = tracker & #dtPending .~ pending'- in (encoded, tracker')+ then Seq.drop 1 pending Seq.|> (seqNum, current)+ else pending Seq.|> (seqNum, current)+ in (encoded, tracker & #dtPending .~ updated) -- | Called when a sequence is ACK'd. -- -- Promotes the matching snapshot to confirmed baseline and discards -- older pending entries. deltaOnAck :: BaselineSeq -> DeltaTracker a -> DeltaTracker a-deltaOnAck seq' tracker =- case Seq.findIndexL (\(s, _) -> s == seq') (dtPending tracker) of+deltaOnAck seqNum tracker =+ case Seq.findIndexL (\(s, _) -> s == seqNum) (dtPending tracker) of Nothing -> tracker Just idx ->- let (ackSeq, snapshot) = Seq.index (dtPending tracker) idx+ case Seq.lookup idx (dtPending tracker) of+ Nothing -> tracker+ Just (ackSeq, snapshot) -> -- Drop everything older than the acked position- pending' =- Seq.filter (\(s, _) -> baselineSeqDiff s ackSeq >= 0) $- Seq.drop (idx + 1) (dtPending tracker)- in tracker- & #dtPending- .~ pending'- & #dtConfirmed- ?~ (ackSeq, snapshot)+ let remaining =+ Seq.filter (\(s, _) -> baselineSeqDiff s ackSeq >= 0) $+ Seq.drop (idx + 1) (dtPending tracker)+ in tracker+ & #dtPending+ .~ remaining+ & #dtConfirmed+ ?~ (ackSeq, snapshot) -- | Reset tracker state (e.g. on reconnect). deltaReset :: DeltaTracker a -> DeltaTracker a@@ -233,22 +233,23 @@ -- | Store a confirmed snapshot at the given sequence. pushBaseline :: BaselineSeq -> a -> MonoTime -> BaselineManager a -> BaselineManager a-pushBaseline seq' state now manager =+pushBaseline seqNum state now manager = let evictExpired = Seq.filter (\(_, _, ts) -> elapsedMs ts now < bmTimeoutMs manager) evictOldest s | Seq.length s >= bmMaxSnapshots manager = Seq.drop 1 s | otherwise = s- snapshots = evictOldest (evictExpired (bmSnapshots manager)) Seq.|> (seq', state, now)+ snapshots = evictOldest (evictExpired (bmSnapshots manager)) Seq.|> (seqNum, state, now) in manager & #bmSnapshots .~ snapshots -- | Look up a baseline by sequence number. getBaseline :: BaselineSeq -> BaselineManager a -> Maybe a-getBaseline seq' manager =- case Seq.findIndexR (\(s, _, _) -> s == seq') (bmSnapshots manager) of+getBaseline seqNum manager =+ case Seq.findIndexR (\(s, _, _) -> s == seqNum) (bmSnapshots manager) of Nothing -> Nothing Just idx ->- let (_, state, _) = Seq.index (bmSnapshots manager) idx- in Just state+ case Seq.lookup idx (bmSnapshots manager) of+ Nothing -> Nothing+ Just (_, state, _) -> Just state -- | Clear all stored baselines. baselineReset :: BaselineManager a -> BaselineManager a
src/GBNet/Replication/Interpolation.hs view
@@ -19,10 +19,10 @@ -- -- @ -- -- On receiving server snapshot--- let buffer' = pushSnapshot serverTimestamp playerState buffer+-- let updated = pushSnapshot serverTimestamp playerState buffer -- -- -- On render tick--- case sampleSnapshot renderTime buffer' of+-- case sampleSnapshot renderTime updated of -- Nothing -> renderLastKnown -- Just interpolated -> render interpolated -- @@@ -137,17 +137,17 @@ EmptyR -> -- First snapshot buffer & #sbSnapshots .~ Seq.singleton (TimestampedSnapshot timestamp state)- _ :> last' ->- if timestamp <= tsTimestamp last'+ _ :> newest ->+ if timestamp <= tsTimestamp newest then buffer -- Drop out-of-order else- let snapshots' = snapshots |> TimestampedSnapshot timestamp state+ let appended = snapshots |> TimestampedSnapshot timestamp state -- Keep buffer bounded maxEntries = sbBufferDepth buffer * 2 trimmed =- if Seq.length snapshots' > maxEntries- then Seq.drop (Seq.length snapshots' - maxEntries) snapshots'- else snapshots'+ if Seq.length appended > maxEntries+ then Seq.drop (Seq.length appended - maxEntries) appended+ else appended in buffer & #sbSnapshots .~ trimmed -- | Sample an interpolated state at @renderTime@ (in milliseconds).
src/GBNet/Replication/Priority.hs view
@@ -22,7 +22,7 @@ -- & register npcId 2.0 -- low priority (2 units/sec) -- -- -- Each tick, accumulate priority based on elapsed time--- let acc' = accumulate 0.016 acc -- 16ms tick+-- let accumulated = accumulate 0.016 acc -- 16ms tick -- -- -- Drain entities that fit in budget -- let (selected, acc'') = drainTop 1200 entitySize acc'@@ -139,12 +139,12 @@ (selected, _remaining) = selectWithinBudget budgetBytes sizeFunc sorted [] -- Reset priority for selected entities- entries' =+ resetEntries = foldr (Map.adjust (#peAccumulated .~ 0.0)) entries selected- in (selected, PriorityAccumulator entries')+ in (selected, PriorityAccumulator resetEntries) -- | Helper to select entities within budget. selectWithinBudget ::
src/GBNet/Security.hs view
@@ -132,12 +132,12 @@ -- Automatically prunes stale entries when the cleanup interval has elapsed. rateLimiterAllow :: Word64 -> MonoTime -> RateLimiter -> (Bool, RateLimiter) rateLimiterAllow addrKey now rl- | recentCount >= rlMaxRequestsPerSecond rl' = (False, rl' & #rlRequests %~ Map.insert addrKey recent)- | otherwise = (True, rl' & #rlRequests %~ Map.insert addrKey (now : recent))+ | recentCount >= rlMaxRequestsPerSecond cleaned = (False, cleaned & #rlRequests %~ Map.insert addrKey recent)+ | otherwise = (True, cleaned & #rlRequests %~ Map.insert addrKey (now : recent)) where- rl' = maybeCleanup now rl- window = rlWindowMs rl'- timestamps = Map.findWithDefault [] addrKey (rlRequests rl')+ cleaned = maybeCleanup now rl+ window = rlWindowMs cleaned+ timestamps = Map.findWithDefault [] addrKey (rlRequests cleaned) -- Filter and count in single pass (recentCount, recent) = foldr countRecent (0, []) timestamps countRecent t (n, acc)@@ -214,10 +214,8 @@ | isTokenExpired now token = (Left TokenExpired, tv) | Map.member (ctClientId token) (tvUsedTokens tv) = (Left TokenReplayed, tv) | otherwise =- let tv' =- tv & #tvUsedTokens %~ Map.insert (ctClientId token) now- tv'' = enforceLimit now tv'- in (Right (ctClientId token), tv'')+ let tracked = tv & #tvUsedTokens %~ Map.insert (ctClientId token) now+ in (Right (ctClientId token), enforceLimit now tracked) -- | Enforce maximum tracked tokens limit. enforceLimit :: MonoTime -> TokenValidator -> TokenValidator
src/GBNet/Socket.hs view
@@ -112,17 +112,19 @@ return $ case result of Left err -> Left (SocketIoError err) Right sent ->- let sock' =- sock- & #usStats- % #ssBytesSent- %~ (+ fromIntegral sent)- & #usStats- % #ssPacketsSent- %~ (+ 1)- & (#usStats % #ssLastSendTime)- ?~ now- in Right (sent, sock')+ Right+ ( sent,+ sock+ & #usStats+ % #ssBytesSent+ %~ (+ fromIntegral sent)+ & #usStats+ % #ssPacketsSent+ %~ (+ 1)+ & #usStats+ % #ssLastSendTime+ ?~ now+ ) -- | Try an IO action, catching IOExceptions. tryIO :: IO a -> IO (Either String a)
src/GBNet/TestNet.hs view
@@ -155,17 +155,17 @@ then pure (Left NetSocketClosed) else do let cfg = tnsConfig st- (r1, rng') = nextRandom (tnsRng st)+ (r1, rng1) = nextRandom (tnsRng st) if randomDouble r1 < tncLossRate cfg then do- #tnsRng .= rng'+ #tnsRng .= rng1 pure (Right ()) else do- let (r2, rng'') = nextRandom rng'+ let (r2, rng2) = nextRandom rng1 jitterRange = tncJitterNs cfg jitter = if jitterRange == 0 then 0 else r2 `mod` (jitterRange + 1) -- Out-of-order: add extra random delay- (r3, rng3) = nextRandom rng''+ (r3, rng3) = nextRandom rng2 oooDelay = if randomDouble r3 < tncOutOfOrderChance cfg then MonoTime (r3 `mod` (testNetOutOfOrderMaxDelayNs + 1))@@ -263,9 +263,8 @@ (initialTestNetState addr) {tnsCurrentTime = twGlobalTime world} addr (twPeers world)- (result, peerState') = runTestNet action peerState- world' = world & #twPeers %~ Map.insert addr peerState'- in (result, world')+ (result, updated) = runTestNet action peerState+ in (result, world & #twPeers %~ Map.insert addr updated) -- | Deliver all ready packets between peers. -- Packets are moved from sender's outFlight to receiver's inbox@@ -301,9 +300,9 @@ -- | Advance time for all peers in the world. worldAdvanceTime :: MonoTime -> TestWorld -> TestWorld worldAdvanceTime newTime world =- let world' = world & #twGlobalTime .~ newTime+ let timed = world & #twGlobalTime .~ newTime updatedPeers = Map.map (\ps -> ps & #tnsCurrentTime .~ newTime)- (twPeers world')- in deliverPackets (world' & #twPeers .~ updatedPeers)+ (twPeers timed)+ in deliverPackets (timed & #twPeers .~ updatedPeers)
src/GBNet/Util.hs view
@@ -1,4 +1,9 @@--- | Shared utilities: sequence number wraparound and deterministic RNG.+-- |+-- Module : GBNet.Util+-- Description : Sequence number wraparound and deterministic RNG+--+-- Shared utilities used across the library: circular sequence number+-- comparison, signed difference, and a pure SplitMix-based RNG. module GBNet.Util ( sequenceHalfRange, sequenceGreaterThan,
src/GBNet/ZeroCopy.hs view
@@ -61,8 +61,8 @@ zeroCopyMutateU' vec action = runST $ do mv <- VU.unsafeThaw vec result <- action mv- v' <- VU.unsafeFreeze mv- return (v', result)+ frozen <- VU.unsafeFreeze mv+ return (frozen, result) {-# INLINE zeroCopyMutateU' #-} -- | Zero-copy mutation of a boxed vector.@@ -89,6 +89,6 @@ zeroCopyMutate' vec action = runST $ do mv <- V.unsafeThaw vec result <- action mv- v' <- V.unsafeFreeze mv- return (v', result)+ frozen <- V.unsafeFreeze mv+ return (frozen, result) {-# INLINE zeroCopyMutate' #-}
test/Main.hs view
@@ -11,6 +11,7 @@ import Data.Bits ((.&.)) import qualified Data.ByteString as BS import Data.List (foldl')+import qualified Data.Map.Strict as Map import Data.Word (Word16, Word32, Word64, Word8) import GBNet.Channel import GBNet.Class ()@@ -56,7 +57,7 @@ import GBNet.Serialize.TH (deriveStorable) import GBNet.Simulator import GBNet.Socket (UdpSocket (..))-import GBNet.Stats (CongestionLevel (..), defaultSocketStats)+import GBNet.Stats (CongestionLevel (..), NetworkStats (..), defaultSocketStats) import GBNet.TestNet import GBNet.Types (ChannelId (..), MessageId (..), SequenceNum (..)) import GBNet.Util@@ -259,6 +260,12 @@ -- IPv6 address helpers testIPv6Helpers + -- Bandwidth tracking+ testBandwidthTracking++ -- Migration cooldown sweep+ testMigrationCooldownSweep+ putStrLn "" putStrLn "All tests passed!" @@ -291,7 +298,7 @@ assertEqual "Vec3 size" 12 (BS.length bytes) case deserialize bytes :: Either String Vec3 of Left err -> error $ " FAIL: deserialize Vec3: " ++ err- Right v' -> assertEqual "Vec3 roundtrip" v v'+ Right decoded -> assertEqual "Vec3 roundtrip" v decoded testPacketHeaderRoundTrip :: IO () testPacketHeaderRoundTrip = do@@ -309,11 +316,11 @@ assertEqual ("size for " ++ show (packetType hdr)) packetHeaderByteSize (BS.length bytes) case deserializeHeader bytes of Left err -> error $ "deserialize failed: " ++ err- Right hdr' -> do- assertEqual "roundtrip packetType" (packetType hdr) (packetType hdr')- assertEqual "roundtrip sequenceNum" (sequenceNum hdr) (sequenceNum hdr')- assertEqual "roundtrip ack" (ack hdr) (ack hdr')- assertEqual "roundtrip ackBitfield" (ackBitfield hdr) (ackBitfield hdr')+ Right decoded -> do+ assertEqual "roundtrip packetType" (packetType hdr) (packetType decoded)+ assertEqual "roundtrip sequenceNum" (sequenceNum hdr) (sequenceNum decoded)+ assertEqual "roundtrip ack" (ack hdr) (ack decoded)+ assertEqual "roundtrip ackBitfield" (ackBitfield hdr) (ackBitfield decoded) -------------------------------------------------------------------------------- -- Reliability module tests@@ -649,8 +656,8 @@ Right _ -> error " FAIL: should have failed" -- Broadcast to empty peer should be no-op- let peer' = peerBroadcast (ChannelId 0) "test" Nothing now peer- assertEqual "broadcast to empty" 0 (peerCount peer')+ let broadcasted = peerBroadcast (ChannelId 0) "test" Nothing now peer+ assertEqual "broadcast to empty" 0 (peerCount broadcasted) putStrLn " PASS: broadcast to empty peer is no-op" testPeerDisconnect :: IO ()@@ -663,8 +670,8 @@ peer = newPeerState sock addr config now -- Disconnect from non-connected peer is no-op- let peer' = peerDisconnect (peerIdFromAddr (testAddr 1111)) now peer- assertEqual "disconnect non-existing" 0 (peerCount peer')+ let disconnected = peerDisconnect (peerIdFromAddr (testAddr 1111)) now peer+ assertEqual "disconnect non-existing" 0 (peerCount disconnected) -- Connect then disconnect let peer1 = peerConnect (peerIdFromAddr (testAddr 2222)) now peer@@ -1003,7 +1010,7 @@ propStorableRoundTrip :: Vec3 -> Bool propStorableRoundTrip v = case deserialize (serialize v) of- Right v' -> v == v'+ Right decoded -> v == decoded Left _ -> False -- | PacketHeader serialize/deserialize roundtrip@@ -1011,21 +1018,21 @@ propPacketHeaderRoundTrip hdr = case deserializeHeader (serializeHeader hdr) of Left _ -> False- Right hdr' ->- packetType hdr == packetType hdr'- && sequenceNum hdr == sequenceNum hdr'- && ack hdr == ack hdr'- && ackBitfield hdr == ackBitfield hdr'+ Right decoded ->+ packetType hdr == packetType decoded+ && sequenceNum hdr == sequenceNum decoded+ && ack hdr == ack decoded+ && ackBitfield hdr == ackBitfield decoded -- | FragmentHeader serialize/deserialize roundtrip propFragmentHeaderRoundTrip :: FragmentHeader -> Bool propFragmentHeaderRoundTrip hdr = case deserializeFragmentHeader (serializeFragmentHeader hdr) of Nothing -> False- Just hdr' ->- fhMessageId hdr == fhMessageId hdr'- && fhFragmentIndex hdr == fhFragmentIndex hdr'- && fhFragmentCount hdr == fhFragmentCount hdr'+ Just decoded ->+ fhMessageId hdr == fhMessageId decoded+ && fhFragmentIndex hdr == fhFragmentIndex decoded+ && fhFragmentCount hdr == fhFragmentCount decoded -- | sequenceGreaterThan is antisymmetric: if a > b then not (b > a) propSeqGtAntisymmetric :: SequenceNum -> SequenceNum -> Bool@@ -1152,76 +1159,78 @@ testConnectionMigration = do putStrLn "Connection migration:" sock <- newTestUdpSocket- let serverAddr = testAddr 7010- clientAddr = testAddr 8010+ let serverAddr = testAddr serverPort+ clientAddr = testAddr clientPort config = defaultNetworkConfig {ncEnableConnectionMigration = True}- startTime = 1000000000 :: MonoTime+ startTime = oneSecondNs - let serverPeer = newPeerState sock serverAddr config 100000000- clientPeer0 = newPeerState sock clientAddr config 200000000+ let serverPeer = newPeerState sock serverAddr config serverSeed+ clientPeer0 = newPeerState sock clientAddr config clientSeed clientPeer1 = peerConnect (peerIdFromAddr serverAddr) startTime clientPeer0 -- Full handshake via TestNet let world0 = initWorld startTime serverAddr clientAddr-- -- Tick 1: client sends request let ((_, cp2), w1) = tickPeerInWorld clientAddr [] clientPeer1 world0- let w2 = stepWorld 10 w1-- -- Tick 2: server sends challenge+ let w2 = stepWorld tickStepMs w1 let ((_, sp1), w3) = tickPeerInWorld serverAddr [] serverPeer w2- let w4 = stepWorld 10 w3-- -- Tick 3: client sends response+ let w4 = stepWorld tickStepMs w3 let ((_, cp3), w5) = tickPeerInWorld clientAddr [] cp2 w4- let w6 = stepWorld 10 w5-- -- Tick 4: server accepts+ let w6 = stepWorld tickStepMs w5 let ((_, sp2), w7) = tickPeerInWorld serverAddr [] sp1 w6- let w8 = stepWorld 10 w7-- -- Tick 5: client receives accepted- let ((_, _cp4), w9) = tickPeerInWorld clientAddr [] cp3 w8- let _w10 = stepWorld 10 w9+ let w8 = stepWorld tickStepMs w7+ let ((_, cp4), w9) = tickPeerInWorld clientAddr [] cp3 w8+ let _w10 = stepWorld tickStepMs w9 -- Verify connection established assertEqual "server has 1 connection" 1 (peerCount sp2) assertEqual "server knows client" True (peerIsConnected (peerIdFromAddr clientAddr) sp2) - -- Now simulate migration: take the outgoing packets from client, but- -- present them to the server as coming from a new address- let newClientAddr = testAddr 8099- newClientPid = peerIdFromAddr newClientAddr-- -- Get a valid packet from the connected client- let clientResult = peerProcess (startTime + 50000000) [] sp2- outgoing = prOutgoing clientResult-- case outgoing of- [] -> do- -- No outgoing packets, craft a minimal valid one by re-processing- assertEqual "migration config enabled" True (ncEnableConnectionMigration config)- putStrLn " PASS: Migration enabled (no packets to migrate with)"- (firstPkt : _) -> do- -- Re-present this packet as coming from the new address- let migratedPkt = IncomingPacket newClientPid (rpData firstPkt)- migrateTime = startTime + 60000000000 -- well past cooldown- result = peerProcess migrateTime [migratedPkt] sp2- events = prEvents result- migrated = [() | PeerMigrated _ _ <- events]+ -- Queue a message on the CLIENT so peerProcess produces Payload packets.+ -- Migration only triggers for Payload type from unknown peers.+ let serverPid = peerIdFromAddr serverAddr+ sendTime = startTime + postHandshakeOffsetNs+ case peerSend serverPid (ChannelId 0) testPayload sendTime cp4 of+ Left err -> error $ " FAIL: peerSend: " ++ show err+ Right clientWithMsg -> do+ -- Get client's outgoing (CRC-wrapped), strip CRC, present from new address+ let clientResult = peerProcess sendTime [] clientWithMsg+ newClientPid = peerIdFromAddr (testAddr migratedPort)+ stripped =+ concatMap+ ( \pkt -> case validateAndStripCrc32 (rpData pkt) of+ Nothing -> []+ Just valid -> [IncomingPacket newClientPid valid]+ )+ (prOutgoing clientResult) - if not (null migrated)- then do- -- Migration fired- assertEqual "old connection gone" False (peerIsConnected (peerIdFromAddr clientAddr) (prPeer result))- assertEqual "new connection exists" True (peerIsConnected newClientPid (prPeer result))- putStrLn " PASS: Connection migration"- else do- -- Packet may have been rejected for other reasons (CRC, sequence distance)+ case stripped of+ [] -> do assertEqual "migration config enabled" True (ncEnableConnectionMigration config)- assertEqual "server still has connection" 1 (peerCount (prPeer result))- putStrLn " PASS: Migration wired up (packet not matched)"+ putStrLn " PASS: Migration enabled (no packets to migrate with)"+ _ -> do+ let result = peerProcess sendTime stripped sp2+ events = prEvents result+ migrated = [() | PeerMigrated _ _ <- events] + if null migrated+ then do+ assertEqual "migration config enabled" True (ncEnableConnectionMigration config)+ putStrLn " PASS: Migration wired up (packet not matched)"+ else do+ assertEqual "old connection gone" False (peerIsConnected (peerIdFromAddr clientAddr) (prPeer result))+ assertEqual "new connection exists" True (peerIsConnected newClientPid (prPeer result))+ putStrLn " PASS: Connection migration"+ where+ serverPort = 7010+ clientPort = 8010+ migratedPort = 8099+ serverSeed = 100000000+ clientSeed = 200000000+ oneSecondNs = 1000000000 :: MonoTime+ tickStepMs = 10 :: MonoTime+ postHandshakeOffsetNs = 100000000 :: MonoTime -- 100ms+ testPayload = "migration-test"+ -------------------------------------------------------------------------------- -- Connection state machine tests --------------------------------------------------------------------------------@@ -1250,14 +1259,14 @@ Right _ -> error " FAIL: double connect should fail" -- createHeader increments local sequence- let conn0' = newConnection config clientSalt now- let seqBefore = connLocalSeq conn0'- let (_header, conn1') = createHeader conn0'- assertEqual "local seq incremented" (seqBefore + 1) (connLocalSeq conn1')+ let connA = newConnection config clientSalt now+ let seqBefore = connLocalSeq connA+ let (_header, connB) = createHeader connA+ assertEqual "local seq incremented" (seqBefore + 1) (connLocalSeq connB) -- Second createHeader increments again- let (_header2, conn2') = createHeader conn1'- assertEqual "local seq incremented again" (seqBefore + 2) (connLocalSeq conn2')+ let (_header2, connC) = createHeader connB+ assertEqual "local seq incremented again" (seqBefore + 2) (connLocalSeq connC) testConnectionSendReceive :: IO () testConnectionSendReceive = do@@ -1376,14 +1385,14 @@ -- With 0% loss and 0 latency, packet should be delivered immediately let testData = "hello" :: BS.ByteString- testAddr' = 42 :: Word64- (immediate, sim1) = simulatorProcessSend testData testAddr' now sim0+ testAddrKey = 42 :: Word64+ (immediate, sim1) = simulatorProcessSend testData testAddrKey now sim0 assertEqual "immediate delivery count" 1 (length immediate) case immediate of [(dat, addr)] -> do assertEqual "delivered data" testData dat- assertEqual "delivered addr" testAddr' addr+ assertEqual "delivered addr" testAddrKey addr _ -> error " FAIL: unexpected immediate result" -- Nothing should be queued since latency is 0@@ -1392,7 +1401,7 @@ -- Test with latency: packets should be delayed let configWithLatency = defaultSimulationConfig {simLatencyMs = 100} sim2 = newNetworkSimulator configWithLatency now- (immediate2, sim3) = simulatorProcessSend testData testAddr' now sim2+ (immediate2, sim3) = simulatorProcessSend testData testAddrKey now sim2 assertEqual "no immediate with latency" 0 (length immediate2) assertEqual "1 pending with latency" 1 (simulatorPendingCount sim3)@@ -1412,7 +1421,7 @@ case lateResults of [(dat, addr)] -> do assertEqual "received data" testData dat- assertEqual "received addr" testAddr' addr+ assertEqual "received addr" testAddrKey addr _ -> error " FAIL: unexpected late result" -- | Test that reliable messages survive loss + latency via Simulator.@@ -1482,27 +1491,27 @@ -- Process client → get outgoing (includes queued message) clientResult = peerProcess tickTime [] client- client' = prPeer clientResult+ nextClient = prPeer clientResult clientOut = prOutgoing clientResult -- Feed client outgoing through C2S Simulator- (sC2S', serverPkts) = conditionPackets clientOut clientAddrKey clientAddr tickTime sC2S+ (nextC2S, serverPkts) = conditionPackets clientOut clientAddrKey clientAddr tickTime sC2S -- Process server with conditioned client packets serverResult = peerProcess tickTime serverPkts server- server' = prPeer serverResult+ nextServer = prPeer serverResult serverOut = prOutgoing serverResult events = prEvents serverResult -- Feed server outgoing through S2C Simulator- (sS2C', clientPkts) = conditionPackets serverOut serverAddrKey serverAddr tickTime sS2C+ (nextS2C, clientPkts) = conditionPackets serverOut serverAddrKey serverAddr tickTime sS2C -- Feed server responses back to client- clientResult2 = peerProcess tickTime clientPkts client'- client'' = prPeer clientResult2+ clientResult2 = peerProcess tickTime clientPkts nextClient+ finalClient = prPeer clientResult2 gotMessage = any isMessage events- in (gotMessage, server', client'', sC2S', sS2C')+ in (gotMessage, nextServer, finalClient, nextC2S, nextS2C) ) (False, sp2, cp5, simC2S0, simS2C0) [1 .. tickCount]@@ -1518,14 +1527,15 @@ conditionPackets :: [RawPacket] -> Word64 -> SockAddr -> MonoTime -> NetworkSimulator -> (NetworkSimulator, [IncomingPacket]) conditionPackets pkts addrKey fromAddr now sim0 =- let (sim1, incoming) =+ let (sim1, revIncoming) = foldl' ( \(!s, !acc) pkt ->- let (immediate, s') = simulatorProcessSend (rpData pkt) addrKey now s- in (s', acc ++ stripAndWrap immediate)+ let (immediate, advanced) = simulatorProcessSend (rpData pkt) addrKey now s+ in (advanced, reverse (stripAndWrap immediate) ++ acc) ) (sim0, []) pkts+ incoming = reverse revIncoming -- Also deliver any previously delayed packets now ready (delayed, sim2) = simulatorReceiveReady now sim1 allIncoming = incoming ++ stripAndWrap delayed@@ -1692,8 +1702,8 @@ let (result, _assembler) = foldl feedFrag (Nothing, assembler0) frags where feedFrag (prevResult, asm) frag =- let (r, asm') = processFragment frag now asm- in (case r of Nothing -> prevResult; Just _ -> r, asm')+ let (r, updated) = processFragment frag now asm+ in (case r of Nothing -> prevResult; Just _ -> r, updated) case result of Nothing -> error " FAIL: reassembly did not produce a result" Just reassembled ->@@ -1953,3 +1963,175 @@ let addr3 = SockAddrInet6 (fromIntegral (9999 :: Word16)) 0 (1, 2, 3, 4) 0 assertEqual "ipv6 tuple" addr3 addr3 putStrLn " PASS: IPv6 helpers produce SockAddrInet6"++--------------------------------------------------------------------------------+-- Bandwidth tracking+--------------------------------------------------------------------------------++testBandwidthTracking :: IO ()+testBandwidthTracking = do+ putStrLn "Bandwidth tracking records bytes sent/received:"+ sock <- newTestUdpSocket+ let serverAddr = testAddr serverPort+ clientAddr = testAddr clientPort+ config = defaultNetworkConfig+ startTime = oneSecondNs++ let serverPeer = newPeerState sock serverAddr config serverSeed+ clientPeer0 = newPeerState sock clientAddr config clientSeed+ clientPeer1 = peerConnect (peerIdFromAddr serverAddr) startTime clientPeer0++ let world0 = initWorld startTime serverAddr clientAddr++ -- Full handshake via TestNet+ let ((_, cp2), w1) = tickPeerInWorld clientAddr [] clientPeer1 world0+ let w2 = stepWorld tickStepMs w1+ let ((_, sp1), w3) = tickPeerInWorld serverAddr [] serverPeer w2+ let w4 = stepWorld tickStepMs w3+ let ((_, cp3), w5) = tickPeerInWorld clientAddr [] cp2 w4+ let w6 = stepWorld tickStepMs w5+ let ((_, sp2), w7) = tickPeerInWorld serverAddr [] sp1 w6+ let w8 = stepWorld tickStepMs w7+ let ((_, cp4), w9) = tickPeerInWorld clientAddr [] cp3 w8+ let w10 = stepWorld tickStepMs w9++ -- Both connected. Client sends a message on channel 0.+ let ((_, cp5), w11) = tickPeerInWorld clientAddr [(ChannelId 0, testPayload)] cp4 w10+ let w12 = stepWorld tickStepMs w11++ -- Server receives the message+ let ((_, sp3), w13) = tickPeerInWorld serverAddr [] sp2 w12+ let w14 = stepWorld tickStepMs w13++ -- Client receives server's ACK/response+ let ((_, cp6), _w15) = tickPeerInWorld clientAddr [] cp5 w14++ -- Client should have recorded bytes sent (message payload via drainAllConnectionQueues)+ let serverPid = peerIdFromAddr serverAddr+ case peerStats serverPid cp6 of+ Nothing -> error " FAIL: client has no stats for server connection"+ Just clientStats -> do+ assertEqual "client nsBytesSent > 0" True (nsBytesSent clientStats > 0)+ assertEqual "client nsPacketsSent > 0" True (nsPacketsSent clientStats > 0)++ -- Server should have recorded bytes received (from client's message)+ let clientPid = peerIdFromAddr clientAddr+ case peerStats clientPid sp3 of+ Nothing -> error " FAIL: server has no stats for client connection"+ Just serverStats -> do+ assertEqual "server nsBytesReceived > 0" True (nsBytesReceived serverStats > 0)+ assertEqual "server nsPacketsReceived > 0" True (nsPacketsReceived serverStats > 0)+ -- Server also sends packets back (keepalive/ACK via drainAllConnectionQueues)+ assertEqual "server nsBytesSent > 0" True (nsBytesSent serverStats > 0)++ putStrLn " PASS: Bandwidth tracking records bytes"+ where+ serverPort = 7020+ clientPort = 8020+ serverSeed = 100000000+ clientSeed = 200000000+ oneSecondNs = 1000000000 :: MonoTime+ tickStepMs = 10 :: MonoTime+ testPayload = "bandwidth-tracking-test"++--------------------------------------------------------------------------------+-- Migration cooldown sweep+--------------------------------------------------------------------------------++testMigrationCooldownSweep :: IO ()+testMigrationCooldownSweep = do+ putStrLn "Migration cooldown sweep removes stale entries:"+ sock <- newTestUdpSocket+ let serverAddr = testAddr serverPort+ clientAddr = testAddr clientPort+ config = defaultNetworkConfig {ncEnableConnectionMigration = True}+ startTime = oneSecondNs++ let serverPeer = newPeerState sock serverAddr config serverSeed+ clientPeer0 = newPeerState sock clientAddr config clientSeed+ clientPeer1 = peerConnect (peerIdFromAddr serverAddr) startTime clientPeer0++ let world0 = initWorld startTime serverAddr clientAddr++ -- Full handshake via TestNet+ let ((_, cp2), w1) = tickPeerInWorld clientAddr [] clientPeer1 world0+ let w2 = stepWorld tickStepMs w1+ let ((_, sp1), w3) = tickPeerInWorld serverAddr [] serverPeer w2+ let w4 = stepWorld tickStepMs w3+ let ((_, cp3), w5) = tickPeerInWorld clientAddr [] cp2 w4+ let w6 = stepWorld tickStepMs w5+ let ((_, sp2), w7) = tickPeerInWorld serverAddr [] sp1 w6+ let w8 = stepWorld tickStepMs w7+ let ((_, cp4), w9) = tickPeerInWorld clientAddr [] cp3 w8+ let _w10 = stepWorld tickStepMs w9++ -- Verify connection established+ assertEqual "server has 1 connection" 1 (peerCount sp2)++ -- Queue a message on the client so peerProcess produces Payload packets.+ -- Short offset stays within connection timeout (10s).+ let serverPid = peerIdFromAddr serverAddr+ sendTime = startTime + postHandshakeOffsetNs+ case peerSend serverPid (ChannelId 0) testPayload sendTime cp4 of+ Left err -> error $ " FAIL: peerSend: " ++ show err+ Right clientWithMsg -> do+ -- Get client's outgoing packets (CRC-wrapped)+ let clientResult = peerProcess sendTime [] clientWithMsg+ outgoing = prOutgoing clientResult++ -- Strip CRC and present as incoming from a new address+ let newClientPid = peerIdFromAddr (testAddr migratedPort)+ stripped =+ concatMap+ ( \pkt -> case validateAndStripCrc32 (rpData pkt) of+ Nothing -> []+ Just valid -> [IncomingPacket newClientPid valid]+ )+ outgoing++ case stripped of+ [] -> putStrLn " PASS: Migration cooldown sweep (no valid packets)"+ _ -> do+ let migrateResult = peerProcess sendTime stripped sp2+ events = prEvents migrateResult+ migrated = [() | PeerMigrated _ _ <- events]++ if null migrated+ then do+ assertEqual "migration config enabled" True (ncEnableConnectionMigration config)+ putStrLn " PASS: Migration cooldown sweep (migration didn't trigger)"+ else do+ let serverAfterMigration = prPeer migrateResult++ -- After migration, cooldowns map should have an entry+ assertEqual+ "cooldowns non-empty after migration"+ True+ (not (Map.null (npMigrationCooldowns serverAfterMigration)))++ -- Advance time past cooldown (5000ms) but within connection+ -- timeout (10000ms).+ let sweepTime = sendTime + pastCooldownNs++ -- Tick to trigger updateConnections which sweeps stale cooldowns+ let sweepResult = peerProcess sweepTime [] serverAfterMigration+ serverAfterSweep = prPeer sweepResult++ -- Stale cooldown entry should be removed+ assertEqual+ "cooldowns empty after sweep"+ True+ (Map.null (npMigrationCooldowns serverAfterSweep))++ putStrLn " PASS: Migration cooldown sweep removes stale entries"+ where+ serverPort = 7030+ clientPort = 8030+ migratedPort = 8199+ serverSeed = 100000000+ clientSeed = 200000000+ oneSecondNs = 1000000000 :: MonoTime+ tickStepMs = 10 :: MonoTime+ postHandshakeOffsetNs = 100000000 :: MonoTime -- 100ms+ pastCooldownNs = 6000000000 :: MonoTime -- 6s (> 5s cooldown, < 10s timeout)+ testPayload = "migrate-me"