diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -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 diff --git a/gbnet-hs.cabal b/gbnet-hs.cabal --- a/gbnet-hs.cabal +++ b/gbnet-hs.cabal @@ -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 diff --git a/src/GBNet/Channel.hs b/src/GBNet/Channel.hs --- a/src/GBNet/Channel.hs +++ b/src/GBNet/Channel.hs @@ -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 diff --git a/src/GBNet/Congestion.hs b/src/GBNet/Congestion.hs --- a/src/GBNet/Congestion.hs +++ b/src/GBNet/Congestion.hs @@ -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 diff --git a/src/GBNet/Connection.hs b/src/GBNet/Connection.hs --- a/src/GBNet/Connection.hs +++ b/src/GBNet/Connection.hs @@ -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 diff --git a/src/GBNet/Fragment.hs b/src/GBNet/Fragment.hs --- a/src/GBNet/Fragment.hs +++ b/src/GBNet/Fragment.hs @@ -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 diff --git a/src/GBNet/Net.hs b/src/GBNet/Net.hs --- a/src/GBNet/Net.hs +++ b/src/GBNet/Net.hs @@ -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 diff --git a/src/GBNet/Peer.hs b/src/GBNet/Peer.hs --- a/src/GBNet/Peer.hs +++ b/src/GBNet/Peer.hs @@ -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 diff --git a/src/GBNet/Peer/Handshake.hs b/src/GBNet/Peer/Handshake.hs --- a/src/GBNet/Peer/Handshake.hs +++ b/src/GBNet/Peer/Handshake.hs @@ -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) diff --git a/src/GBNet/Peer/Internal.hs b/src/GBNet/Peer/Internal.hs --- a/src/GBNet/Peer/Internal.hs +++ b/src/GBNet/Peer/Internal.hs @@ -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 diff --git a/src/GBNet/Reliability.hs b/src/GBNet/Reliability.hs --- a/src/GBNet/Reliability.hs +++ b/src/GBNet/Reliability.hs @@ -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 diff --git a/src/GBNet/Replication/Delta.hs b/src/GBNet/Replication/Delta.hs --- a/src/GBNet/Replication/Delta.hs +++ b/src/GBNet/Replication/Delta.hs @@ -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 diff --git a/src/GBNet/Replication/Interpolation.hs b/src/GBNet/Replication/Interpolation.hs --- a/src/GBNet/Replication/Interpolation.hs +++ b/src/GBNet/Replication/Interpolation.hs @@ -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). diff --git a/src/GBNet/Replication/Priority.hs b/src/GBNet/Replication/Priority.hs --- a/src/GBNet/Replication/Priority.hs +++ b/src/GBNet/Replication/Priority.hs @@ -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 :: diff --git a/src/GBNet/Security.hs b/src/GBNet/Security.hs --- a/src/GBNet/Security.hs +++ b/src/GBNet/Security.hs @@ -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 diff --git a/src/GBNet/Socket.hs b/src/GBNet/Socket.hs --- a/src/GBNet/Socket.hs +++ b/src/GBNet/Socket.hs @@ -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) diff --git a/src/GBNet/TestNet.hs b/src/GBNet/TestNet.hs --- a/src/GBNet/TestNet.hs +++ b/src/GBNet/TestNet.hs @@ -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) diff --git a/src/GBNet/Util.hs b/src/GBNet/Util.hs --- a/src/GBNet/Util.hs +++ b/src/GBNet/Util.hs @@ -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, diff --git a/src/GBNet/ZeroCopy.hs b/src/GBNet/ZeroCopy.hs --- a/src/GBNet/ZeroCopy.hs +++ b/src/GBNet/ZeroCopy.hs @@ -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' #-} diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -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"