diff --git a/Sources/SwiftNetwork/QUIC/Recovery.swift b/Sources/SwiftNetwork/QUIC/Recovery.swift index 4ccc697..5db8fd4 100644 --- a/Sources/SwiftNetwork/QUIC/Recovery.swift +++ b/Sources/SwiftNetwork/QUIC/Recovery.swift @@ -363,7 +363,11 @@ struct Recovery: ~Copyable, PrefixedLoggable, NonCopyableTimerUser { ) { if sentEntry.lostTime == .zero && sentEntry.packet.isInFlightEligible { if sentEntry.packet.isAckEliciting { - ackElicitingPacketsInFlight -= 1 + if ackElicitingPacketsInFlight > 0 { + ackElicitingPacketsInFlight -= 1 + } else { + log.fault("Cannot decrement ackElicitingPacketsInFlight below zero") + } let number = sentEntry.packet.number log.datapath( "Ack eliciting packet \(number) acked, decrementing ackElicitingPacketsInFlight to: \(ackElicitingPacketsInFlight)" @@ -656,14 +660,16 @@ struct Recovery: ~Copyable, PrefixedLoggable, NonCopyableTimerUser { _ packets: consuming NetworkUniqueDeque, connection: QUICConnection ) -> Bool { - var packets = packets - guard !packets.isEmpty else { + if packets.isEmpty { return false } + + var packets = packets while !packets.isEmpty { - let packet = packets.remove(at: 0) + let packet = packets.removeFirst() sentPacket(packet, time: connection.now, connection: connection) } + return true } } @@ -910,7 +916,11 @@ struct Recovery: ~Copyable, PrefixedLoggable, NonCopyableTimerUser { // and set the lost time. entry.lostTime = timeNow if entry.packet.isAckEliciting { - ackElicitingPacketsInFlight -= 1 + if ackElicitingPacketsInFlight > 0 { + ackElicitingPacketsInFlight -= 1 + } else { + connection.log.fault("Cannot decrement ackElicitingPacketsInFlight below zero") + } } lostPackets.append(entry.packet.identifier) Recovery.logAckElicitingPacketsInFlight( @@ -1114,42 +1124,41 @@ struct Recovery: ~Copyable, PrefixedLoggable, NonCopyableTimerUser { mutating func sendPTO(connection: QUICConnection, path: QUICPath) { var sentPTO = false - let (_, pnSpace) = getEarliestTime( - earliestTimeType: EarliestTimeType.lastSentAckElicitingTime, - connection: connection - ) - var hasAckEliciting = false - connection.withPendingItems(for: pnSpace) { pendingItems in - hasAckEliciting = pendingItems.hasAckElicitingPendingItems - } - let peerCompletedValidation = peerCompletedValidation(connection: connection) - var shouldClearTimer = false + var discardInitialRecoveryState = false - var hadAckElicitingInFlight = false applyToAllInnerStatesMutable { innerState, packetNumberSpace in let ackElicitingPacketsInFlight = innerState.ackElicitingPacketsInFlight - if ackElicitingPacketsInFlight == 0 { + guard ackElicitingPacketsInFlight > 0 else { return } - hadAckElicitingInFlight = true + connection.log.datapath( "PTO \(path.recoveryState.PTOCount) (\(packetNumberSpace)) fired on path \(path.identifier) with \(ackElicitingPacketsInFlight) ack-eliciting packets in flight" ) + + let hasAckEliciting = connection.withPendingItems(for: packetNumberSpace) { + $0.hasAckElicitingPendingItems + } + if hasAckEliciting { connection.log.datapath("Sending next frames with new data as PTOs") - sentPTO = true let packets = connection.sendFramesFromRecovery( on: path, ignoreCongestionWindow: true, discardInitialRecoveryState: &discardInitialRecoveryState ) - if !innerState.recordSentPackets(packets, connection: connection) { + // Only a recorded packet counts as a probe; the pending items may write no payload. + if innerState.recordSentPackets(packets, connection: connection) { + sentPTO = true + } else { connection.log.datapath( "Unable to force send PTOs, likely flow-controlled or unavailable" ) } - } else if ackElicitingPacketsInFlight > 0 { + + } else { connection.log.datapath("Retransmitting two tail-packets as PTO") + var addedPackets = 0 let packetCount = innerState.outstandingPackets.count for i in 0.. 0 (so the PTO stays armed) while + // being unrebuildable once the flow is closed. + var packet = SentPacketRecord() + packet.identifier = .init(space: .applicationData, number: 0) + packet.isInFlightEligible = true + packet.isAckEliciting = true + packet.totalLength = 20 + 96 + packet.sentPath = connection.currentPath?.identifier ?? .none + packet.transmittedItems.sentStreams.append( + TransmittedItems.SentStream( + flowID: stream.identifier, + streamID: QUICStreamID(0), + offset: 0, + length: 32, + isFinal: true + ) + ) + XCTAssertTrue(packet.transmittedItems.hasRetransmissibleItems) + sentPacket(packet, connection: connection) + + connection.recovery.withImmutableInnerState(packetNumberSpace: .applicationData) { innerState in + XCTAssertEqual(innerState.ackElicitingPacketsInFlight, 1) + } + + // Establish (address-validated) connection: peerCompletedValidation must be true. This is + // the condition under which the anti-deadlock PING is incorrectly skipped. + connection.recovery.received1RTTAck = true + XCTAssertTrue(connection.recovery.peerCompletedValidation(connection: connection)) + XCTAssertEqual(path.recoveryState.PTOCount, 0) + + // Fire the PTO with no new ack-eliciting data pending. + let expectation = XCTestExpectation() + self.connection.context.async { + self.connection.withCurrentPath { path in + self.connection.recovery.sendPTO(connection: self.connection, path: path) + } + expectation.fulfill() + } + wait(for: [expectation], timeout: 5.0) + + // A probe must have been recorded, taking the packets in flight to two, and the PTO counted. + XCTAssertEqual( + connection.recovery.totalAckElicitingPacketsInFlight, + 2, + "PTO produced no probe for a closed-flow tail packet" + ) + XCTAssertEqual(path.recoveryState.PTOCount, 1) + } + + // `sendPTO` must emit a probe; otherwise the PTO makes no progress. A pending item whose flow was + // torn down writes no payload, so ensure the probe only counts once the packet is recorded. + func testPTOProbesWhenNewDataProducesNothing() { + // A stream queued for service whose flow has since been torn down: it is absent from + // `multiplexedFlows`, so writing it produces no payload. + let unregisteredStream = QUICStreamInstance(parent: connection, inbound: true) + unregisteredStream.setup(streamID: QUICStreamID(0), logPrefixer: recoveryTestsLogPrefixer) + XCTAssertNil(connection.flow(for: unregisteredStream.identifier)) + connection.withPendingItems(for: .initial) { pendingItems in + pendingItems.streamsToService.append(unregisteredStream.identifier) + pendingItems.stream = true + } + + // Recovery still sees new ack-eliciting data, so the PTO sends that rather than retransmitting. + let hasPendingAckEliciting = connection.withPendingItems(for: .initial) { + $0.hasAckElicitingPendingItems + } + XCTAssertTrue(hasPendingAckEliciting) + + // One ack-eliciting packet outstanding, so the PTO is armed and the per-space loop runs. + var packet = SentPacketRecord() + packet.identifier = .init(space: .initial, number: 0) + packet.isInFlightEligible = true + packet.isAckEliciting = true + packet.totalLength = 20 + 96 + packet.sentPath = connection.currentPath?.identifier ?? .none + + sentPacket(packet, connection: connection) + + connection.recovery.withImmutableInnerState(packetNumberSpace: .initial) { innerState in + XCTAssertEqual(innerState.ackElicitingPacketsInFlight, 1) + } + + let expectation = XCTestExpectation() + self.connection.context.async { + self.connection.withCurrentPath { path in + self.connection.recovery.sendPTO(connection: self.connection, path: path) + } + expectation.fulfill() + } + wait(for: [expectation], timeout: 5.0) + + // Ensure a probe has been recorded, taking the packets in flight to two. + connection.recovery.withImmutableInnerState(packetNumberSpace: .initial) { innerState in + XCTAssertEqual( + innerState.ackElicitingPacketsInFlight, + 2, + "PTO reported a probe but no packet was sent" + ) + } + } } #endif