diff --git a/FptnLib/fptn b/FptnLib/fptn index fd56bb8..13acece 160000 --- a/FptnLib/fptn +++ b/FptnLib/fptn @@ -1 +1 @@ -Subproject commit fd56bb88f3c7cb1ea3c84243c9aa0448209384d9 +Subproject commit 13acecee048f2f9d80ffede884db16e8d51982d8 diff --git a/FptnLib/src/websocket/WrapperWebsocketClientBridge.cpp b/FptnLib/src/websocket/WrapperWebsocketClientBridge.cpp index 0a3a8b3..d1eb539 100644 --- a/FptnLib/src/websocket/WrapperWebsocketClientBridge.cpp +++ b/FptnLib/src/websocket/WrapperWebsocketClientBridge.cpp @@ -445,18 +445,28 @@ bool WebsocketSwiftBridge::stop() { return active_client != nullptr; } -bool WebsocketSwiftBridge::sendPacket(const uint8_t* packet_data, uint32_t length) { - if (wrapper_ && wrapper_->client && wrapper_->running) { - try { - fptn::common::network::IPPacketData buffer(packet_data, packet_data + length); - auto packet = fptn::common::network::IPPacket::Parse(std::move(buffer)); - if (packet) { - return wrapper_->client->Send(std::move(packet)); - } - } catch (...) { +// PR1B: returns typed send result as uint8_t. +// 0=accepted, 1=queue_full, 2=transport_stopped, 3=invalid_packet. +std::uint8_t WebsocketSwiftBridge::sendPacket(const uint8_t* packet_data, uint32_t length) { + if (!wrapper_ || !wrapper_->running) { + return static_cast(fptn::protocol::https::SendResult::transport_stopped); + } + if (!wrapper_->client) { + return static_cast(fptn::protocol::https::SendResult::transport_stopped); + } + if (!packet_data || length == 0) { + return static_cast(fptn::protocol::https::SendResult::invalid_packet); + } + try { + fptn::common::network::IPPacketData buffer(packet_data, packet_data + length); + auto packet = fptn::common::network::IPPacket::Parse(std::move(buffer)); + if (!packet) { + return static_cast(fptn::protocol::https::SendResult::invalid_packet); } + return static_cast(wrapper_->client->Send(std::move(packet))); + } catch (...) { + return static_cast(fptn::protocol::https::SendResult::invalid_packet); } - return false; } bool WebsocketSwiftBridge::isStarted() const { @@ -464,7 +474,7 @@ bool WebsocketSwiftBridge::isStarted() const { } WebsocketClientBridgeStatus WebsocketSwiftBridge::getStatus() const { - WebsocketClientBridgeStatus status = {false, false, 0, "", "", 0, 0, 0, 0, 0, 0, 0, false, 0, 0, 0, 0, 0, 0, 0, 0}; + WebsocketClientBridgeStatus status = {false, false, 0, "", "", 0, 0, 0, 0, 0, 0, 0, false, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0}; if (!wrapper_) { status.last_error = "Invalid handle"; return status; @@ -491,6 +501,10 @@ WebsocketClientBridgeStatus WebsocketSwiftBridge::getStatus() const { status.effective_rcvbuf_bytes = wrapper_->client->GetEffectiveRcvbufBytes(); status.effective_sndbuf_bytes = wrapper_->client->GetEffectiveSndbufBytes(); status.socket_buffer_set_error_count = wrapper_->client->GetSocketBufferSetErrorCount(); + status.queued_packets = wrapper_->client->GetQueuedPackets(); + status.queued_bytes = wrapper_->client->GetQueuedBytes(); + status.queued_bytes_peak = wrapper_->client->GetQueuedBytesPeak(); + status.queue_full_count = wrapper_->client->GetQueueFullCount(); } status.live_clients = fptn::protocol::https::WebsocketClient::GetLiveClients(); status.active_reader_coroutines = fptn::protocol::https::WebsocketClient::GetActiveReaderCoroutines(); diff --git a/FptnLib/src/websocket/WrapperWebsocketClientBridge.h b/FptnLib/src/websocket/WrapperWebsocketClientBridge.h index 3973639..d00eecf 100644 --- a/FptnLib/src/websocket/WrapperWebsocketClientBridge.h +++ b/FptnLib/src/websocket/WrapperWebsocketClientBridge.h @@ -9,6 +9,7 @@ Distributed under the MIT License (https://opensource.org/licenses/MIT) #include #include +#include #include #include @@ -38,6 +39,11 @@ struct WebsocketClientBridgeStatus { int active_reader_coroutines; int active_sender_coroutines; int socket_buffer_set_error_count; + // PR1B: outbound queue diagnostics. + uint64_t queued_packets; + uint64_t queued_bytes; + uint64_t queued_bytes_peak; + uint64_t queue_full_count; }; // PR0: SWIFT_NONCOPYABLE makes Swift import this as ~Copyable, @@ -73,7 +79,8 @@ class SWIFT_NONCOPYABLE WebsocketSwiftBridge { bool start(); bool stop(); - bool sendPacket(const uint8_t* packet_data, uint32_t length); + // PR1B: returns 0=accepted, 1=queue_full, 2=transport_stopped, 3=invalid_packet. + std::uint8_t sendPacket(const std::uint8_t* packet_data, std::uint32_t length); bool isStarted() const; WebsocketClientBridgeStatus getStatus() const; void registerIPAssignedCallback(IPAssignedCallback callback); diff --git a/FptnVPN/Cpp/Wrappers/WebScoketClientBridge.swift b/FptnVPN/Cpp/Wrappers/WebScoketClientBridge.swift index 2c59831..7085550 100644 --- a/FptnVPN/Cpp/Wrappers/WebScoketClientBridge.swift +++ b/FptnVPN/Cpp/Wrappers/WebScoketClientBridge.swift @@ -29,6 +29,11 @@ struct WebsocketClientStatus: Sendable { let activeReaderCoroutines: Int let activeSenderCoroutines: Int let socketBufferSetErrorCount: Int + // PR1B: outbound queue diagnostics. + let queuedPackets: UInt64 + let queuedBytes: UInt64 + let queuedBytesPeak: UInt64 + let queueFullCount: UInt64 } final class WebsocketClientBridge { @@ -129,14 +134,19 @@ final class WebsocketClientBridge { return ok } - @discardableResult - func sendPacket(_ data: Data) -> Bool { - return data.withUnsafeBytes { buf in - guard let ptr = buf.baseAddress?.assumingMemoryBound(to: UInt8.self) else { - return false + // PR1B: returns typed send result for provider-level diagnostics. + func sendPacket(_ data: Data) -> WebsocketSendResult { + guard !data.isEmpty else { + return .invalidPacket + } + let rawValue: UInt8 = data.withUnsafeBytes { buffer in + guard let pointer = buffer.baseAddress? + .assumingMemoryBound(to: UInt8.self) else { + return WebsocketSendResult.invalidPacket.rawValue } - return clientBridge.sendPacket(ptr, UInt32(data.count)) + return clientBridge.sendPacket(pointer, UInt32(data.count)) } + return WebsocketSendResult(bridgeValue: rawValue) } var isStarted: Bool { @@ -166,7 +176,11 @@ final class WebsocketClientBridge { liveClients: Int(raw.live_clients), activeReaderCoroutines: Int(raw.active_reader_coroutines), activeSenderCoroutines: Int(raw.active_sender_coroutines), - socketBufferSetErrorCount: Int(raw.socket_buffer_set_error_count) + socketBufferSetErrorCount: Int(raw.socket_buffer_set_error_count), + queuedPackets: raw.queued_packets, + queuedBytes: raw.queued_bytes, + queuedBytesPeak: raw.queued_bytes_peak, + queueFullCount: raw.queue_full_count ) } } diff --git a/FptnVPNTests/FptnVPNTests.swift b/FptnVPNTests/FptnVPNTests.swift index f24c307..0eb62e0 100644 --- a/FptnVPNTests/FptnVPNTests.swift +++ b/FptnVPNTests/FptnVPNTests.swift @@ -453,6 +453,46 @@ struct FptnVPNTests { #expect(winner.accessToken == "token-c") } + // PR1B: WebsocketSendResult enum mapping from C++ bridge uint8_t. + @Test func sendResultMapsBridgeValues() { + #expect(WebsocketSendResult(bridgeValue: 0) == .accepted) + #expect(WebsocketSendResult(bridgeValue: 1) == .queueFull) + #expect(WebsocketSendResult(bridgeValue: 2) == .transportStopped) + #expect(WebsocketSendResult(bridgeValue: 3) == .invalidPacket) + #expect(WebsocketSendResult(bridgeValue: 99) == .unknown) + #expect(WebsocketSendResult(bridgeValue: 255) == .unknown) + } + + // PR1B: only .queueFull should feed backpressure. .transportStopped + // breaks the batch. .invalidPacket is counted but does not slow + // valid traffic. This test documents the policy; the C++ level + // accounting tests (byte reservation CAS, rollback, gauge + // lifecycle, batch bounds) require fptn gtest infrastructure. + @Test func sendResultBackpressurePolicy() { + let queueFull: WebsocketSendResult = .queueFull + let stopped: WebsocketSendResult = .transportStopped + let invalid: WebsocketSendResult = .invalidPacket + + var sendFailures: Int64 = 0 + var invalidPackets: Int64 = 0 + var transportStopped = false + + for result in [queueFull, queueFull, stopped, invalid] { + switch result { + case .accepted: break + case .queueFull: sendFailures += 1 + case .transportStopped: + transportStopped = true + case .invalidPacket, .unknown: + invalidPackets += 1 + } + } + + #expect(sendFailures == 2) + #expect(transportStopped == true) + #expect(invalidPackets == 1) + } + private func temporaryDiagnosticsDirectory() throws -> URL { let url = FileManager.default.temporaryDirectory .appendingPathComponent("FptnVPNTests-\(UUID().uuidString)", isDirectory: true) diff --git a/FptnVPNTunnel/Cpp/WebScoketClientBridge.swift b/FptnVPNTunnel/Cpp/WebScoketClientBridge.swift index ac5439d..24029d9 100644 --- a/FptnVPNTunnel/Cpp/WebScoketClientBridge.swift +++ b/FptnVPNTunnel/Cpp/WebScoketClientBridge.swift @@ -29,6 +29,11 @@ struct WebsocketClientStatus: Sendable { let activeReaderCoroutines: Int let activeSenderCoroutines: Int let socketBufferSetErrorCount: Int + // PR1B: outbound queue diagnostics. + let queuedPackets: UInt64 + let queuedBytes: UInt64 + let queuedBytesPeak: UInt64 + let queueFullCount: UInt64 } final class WebsocketClientBridge { @@ -142,14 +147,19 @@ final class WebsocketClientBridge { return ok } - @discardableResult - func sendPacket(_ data: Data) -> Bool { - return data.withUnsafeBytes { buf in - guard let ptr = buf.baseAddress?.assumingMemoryBound(to: UInt8.self) else { - return false + // PR1B: returns typed send result for provider-level diagnostics. + func sendPacket(_ data: Data) -> WebsocketSendResult { + guard !data.isEmpty else { + return .invalidPacket + } + let rawValue: UInt8 = data.withUnsafeBytes { buffer in + guard let pointer = buffer.baseAddress? + .assumingMemoryBound(to: UInt8.self) else { + return WebsocketSendResult.invalidPacket.rawValue } - return clientBridge.sendPacket(ptr, UInt32(data.count)) + return clientBridge.sendPacket(pointer, UInt32(data.count)) } + return WebsocketSendResult(bridgeValue: rawValue) } var isStarted: Bool { @@ -179,7 +189,11 @@ final class WebsocketClientBridge { liveClients: Int(raw.live_clients), activeReaderCoroutines: Int(raw.active_reader_coroutines), activeSenderCoroutines: Int(raw.active_sender_coroutines), - socketBufferSetErrorCount: Int(raw.socket_buffer_set_error_count) + socketBufferSetErrorCount: Int(raw.socket_buffer_set_error_count), + queuedPackets: raw.queued_packets, + queuedBytes: raw.queued_bytes, + queuedBytesPeak: raw.queued_bytes_peak, + queueFullCount: raw.queue_full_count ) } } diff --git a/FptnVPNTunnel/PacketTunnelProvider.swift b/FptnVPNTunnel/PacketTunnelProvider.swift index ae80b55..65dfc5c 100644 --- a/FptnVPNTunnel/PacketTunnelProvider.swift +++ b/FptnVPNTunnel/PacketTunnelProvider.swift @@ -903,22 +903,48 @@ final class PacketTunnelProvider: NEPacketTunnelProvider, @unchecked Sendable { guard self.shouldReadOutboundPackets() else { return } - var totalBytes: Int64 = 0 - var sendFailures: Int64 = 0 + var queueFullPackets: Int64 = 0 + var invalidPackets: Int64 = 0 + var unknownResults: Int64 = 0 + var transportStopped = false let client = self.currentWebSocketClient() - for packet in packets { - totalBytes += Int64(packet.count) - if client?.sendPacket(packet) != true { - sendFailures += 1 + // PR1B: compute total bytes for the complete read batch + // before sending, so packetFlowReadPackets/Bytes both + // describe what was read from NEPacketTunnelFlow. + let packetCount = Int64(packets.count) + let totalBytes = packets.reduce(into: Int64.zero) { + $0 += Int64($1.count) + } + + // PR1B: use typed send result. Only .queueFull feeds + // backpressure. .transportStopped breaks the batch early. + // .invalidPacket/.unknown are counted separately. + packetLoop: for packet in packets { + switch client?.sendPacket(packet) ?? .transportStopped { + case .accepted: + break + case .queueFull: + queueFullPackets += 1 + case .transportStopped: + transportStopped = true + break packetLoop + case .invalidPacket: + invalidPackets += 1 + case .unknown: + unknownResults += 1 } } let backpressureDelay = self.recordPacketFlowRead( - packetCount: Int64(packets.count), + packetCount: packetCount, byteCount: totalBytes, - sendFailures: sendFailures + sendFailures: queueFullPackets ) + + if transportStopped { + return + } if let backpressureDelay { self.scheduleReadLoopAfterBackpressure(delay: backpressureDelay) return diff --git a/SharedTunnelRuntime/TunnelLifecycleRuntime.swift b/SharedTunnelRuntime/TunnelLifecycleRuntime.swift index 0e6d872..f89944a 100644 --- a/SharedTunnelRuntime/TunnelLifecycleRuntime.swift +++ b/SharedTunnelRuntime/TunnelLifecycleRuntime.swift @@ -193,10 +193,23 @@ struct TunnelLifecycleRuntime: Equatable, Sendable { } } +// PR1B: typed send result for the outbound queue. +enum WebsocketSendResult: UInt8, Sendable { + case accepted = 0 + case queueFull = 1 + case transportStopped = 2 + case invalidPacket = 3 + case unknown = 255 + + init(bridgeValue: UInt8) { + self = Self(rawValue: bridgeValue) ?? .unknown + } +} + protocol TunnelWebSocketTransport: AnyObject { func start() -> Bool func stop() -> Bool - func sendPacket(_ data: Data) -> Bool + func sendPacket(_ data: Data) -> WebsocketSendResult } protocol ReconnectScheduling: AnyObject {