test (29a3b46b) · Commits · jan.koester / libnetplus

src/quic.cpp

+39 −13

Original line number	Diff line number	Diff line
		@@ -4945,6 +4945,12 @@ void quic::flushFlowControl() {
		flushBatch();
		}

		bool quic::isStreamSendStopped(uint64_t stream_id) const {
		std::lock_guard<std::recursive_mutex> lock(quic_mtx());
		auto it = _streams.find(stream_id);
		return it != _streams.end() && it->second.send_stopped;
		}

		bool quic::isStreamReset(uint64_t stream_id) const {
		std::lock_guard<std::recursive_mutex> lock(quic_mtx());

		@@ -5102,6 +5108,7 @@ void quic::processFrame(const uint8_t* data, size_t len, size_t& offset, bool& a
		uint64_t sid, app_error;
		if (!readVarInt(sid) \|\| !readVarInt(app_error)) return;
		auto it = _streams.find(sid);
		if (it != _streams.end()) it->second.send_stopped = true;
		if (it != _streams.end() && !it->second.send_fin) {
		// RFC 9000 §3.5: reset our send side in response so the
		// peer isn't left waiting on a response it already told
		@@ -5109,9 +5116,17 @@ void quic::processFrame(const uint8_t* data, size_t len, size_t& offset, bool& a
		// the peer's stream-count credit) happen promptly
		// instead of only once our send finishes naturally,
		// which for an abandoned/redundant fetch may be slow or
		// never happen at all. resetStream() retires this side
		// itself once both directions are terminal.
		resetStream(sid, app_error);
		// never happen at all.
		//
		// Not resetStream(): that also retires the stream once both
		// directions are terminal -- and a server that sends
		// STOP_SENDING after answering early (RFC 9114 §4.1.2) has
		// typically already delivered the whole response, so retiring
		// here erased it before the application could read it. The
		// stream retires on the app's closeStream() instead, like any
		// other completed stream.
		sendResetStreamFrame(sid, app_error, it->second.send_offset);
		it->second.send_fin = true;
		}
		}
		break;
		@@ -7998,6 +8013,14 @@ size_t quic::sendStreamData(uint64_t stream_id, const uint8_t* data, size_t len,
		return 0;
		}

		// The peer told us to stop (STOP_SENDING) and our send side is reset: fail at once
		// rather than stalling on credit that will never come.
		if (auto sit = _streams.find(stream_id); sit != _streams.end() && sit->second.send_stopped) {
		NetException e;
		e[NetException::Error] << "sendStreamData: peer sent STOP_SENDING on stream " << stream_id;
		throw e;
		}

		// Get or create stream
		auto& stream = _streams[stream_id];
		stream.stream_id = stream_id;
		@@ -9028,7 +9051,20 @@ void quic::resetStream(uint64_t stream_id, uint64_t error_code) {

		auto it = _streams.find(stream_id);
		uint64_t final_size = (it != _streams.end()) ? it->second.send_offset : 0;
		sendResetStreamFrame(stream_id, error_code, final_size);

		// No further STREAM frames should follow an abort.
		if (it != _streams.end()) {
		it->second.send_fin = true;
		// Already have a valid iterator from the lookup above — pass it
		// directly instead of making retireStreamIfDone() re-find stream_id.
		retireStreamIfDone(stream_id, it);
		} else {
		retireStreamIfDone(stream_id);
		}
		}

		void quic::sendResetStreamFrame(uint64_t stream_id, uint64_t error_code, uint64_t final_size) {
		std::vector<uint8_t> frame;
		frame.push_back(0x04); // RESET_STREAM frame type
		uint8_t buf[8];
		@@ -9047,16 +9083,6 @@ void quic::resetStream(uint64_t stream_id, uint64_t error_code) {
		// caller typically holds quic_mtx().
		sendPacket(packet.data(), packet.size());
		}

		// No further STREAM frames should follow an abort.
		if (it != _streams.end()) {
		it->second.send_fin = true;
		// Already have a valid iterator from the lookup above — pass it
		// directly instead of making retireStreamIfDone() re-find stream_id.
		retireStreamIfDone(stream_id, it);
		} else {
		retireStreamIfDone(stream_id);
		}
		}

		// ============================================================================

src/socket.h

+8 −0

Original line number	Diff line number	Diff line
		@@ -742,6 +742,7 @@ namespace netplus {
		uint64_t recv_notified_offset = 0; // contiguous bytes already announced to the callback (pull dispatch only)
		bool recv_complete = false; // all data up to FIN received
		bool send_fin = false;
		bool send_stopped = false; // peer sent STOP_SENDING: our send side was reset for it
		bool reset = false;
		// Absolute stream offset corresponding to recv_buffer[0]. Stays
		// 0 outside incremental dispatch (recv_buffer then holds the
		@@ -885,6 +886,11 @@ namespace netplus {
		bool hasStreamData(uint64_t stream_id) const;
		bool isStreamComplete(uint64_t stream_id) const;
		bool isStreamReset(uint64_t stream_id) const;
		// True once the peer asked us to stop sending on this stream (STOP_SENDING) -- e.g. a
		// server that answered before reading the whole request body. The send side is reset
		// at that point (further sendStreamData() calls throw), but whatever the peer sent us
		// stays readable until closeStream().
		bool isStreamSendStopped(uint64_t stream_id) const;

		virtual std::string selectStreamProtocol(const std::string &proto) { return ""; }

		@@ -1202,6 +1208,8 @@ namespace netplus {
		// public stopSending() (which checks first) and closeStream() (which
		// already knows the stream exists from its own lookup).
		void sendStopSendingFrame(uint64_t stream_id, uint64_t error_code);
		// RESET_STREAM on the wire only -- no stream bookkeeping (see resetStream()).
		void sendResetStreamFrame(uint64_t stream_id, uint64_t error_code, uint64_t final_size);

		// Loss detection and retransmission
		void checkLossAndRetransmit();