fin added (e59796bf) · Commits · jan.koester / libnetplus

src/quic.cpp

+78 −10

Original line number	Diff line number	Diff line
		@@ -2944,6 +2944,7 @@ void quic::processFrame(const uint8_t* data, size_t len, size_t& offset) {
		it->second.recv_fin_offset = final_size;
		}
		(void)app_error;
		retireStreamIfDone(sid);
		}
		break;

		@@ -2956,9 +2957,15 @@ void quic::processFrame(const uint8_t* data, size_t len, size_t& offset) {
		offset += bytes;
		auto it = _streams.find(sid);
		if (it != _streams.end()) {
		it->second.send_fin = true;
		// RFC 9000 §3.5: reset the stream in response so the
		// peer can see our send side as done and retire it
		// promptly, instead of waiting on a response it already
		// told us it doesn't want.
		if (!it->second.send_fin) {
		resetStream(sid, app_error);
		}
		retireStreamIfDone(sid);
		}
		(void)app_error;
		}
		break;

		@@ -3569,6 +3576,15 @@ void quic::processStreamFrame(const uint8_t* data, size_t len, size_t& offset) {
		stream.recv_ranges.clear();
		}
		}

		// The peer's FIN just arrived on this stream. If our own side already
		// finished sending (send_fin set earlier via closeStream/resetStream),
		// this is the event that completes the pair — retire it now instead of
		// relying on the local application to call closeStream() again for a
		// stream it may already consider done.
		if (has_fin) {
		retireStreamIfDone(stream_id);
		}
		}

		// ============================================================================
		@@ -5413,6 +5429,28 @@ void quic::closeStream(uint64_t stream_id) {
		stream.send_fin = true;
		}

		// The caller is closing this stream even though the peer's side hasn't
		// finished (no FIN/RESET received yet) — e.g. abandoning a redundant
		// erasure-coded fetch once enough other shards already arrived. Without
		// telling the peer, it has no reason to hurry its send, so its own
		// retirement (and the MAX_STREAMS credit that depends on it) may not
		// happen for a long time, or ever. STOP_SENDING asks it to abort now.
		if (!stream.recv_fin) {
		stopSending(stream_id);
		}

		retireStreamIfDone(stream_id);
		}

		// If a stream has terminated in both directions, erase it and — for a
		// peer-initiated stream — grant the peer fresh stream-count credit.
		// Assumes quic_mtx() is already held by the caller (recursive mutex, so
		// re-entrant calls from within this function are safe too).
		void quic::retireStreamIfDone(uint64_t stream_id) {
		auto it = _streams.find(stream_id);
		if (it == _streams.end()) return;
		Stream& stream = it->second;

		// If both sides have sent FIN, remove stream and replenish peer's budget
		if (stream.recv_fin && stream.send_fin) {
		bool peer_bidi = _is_server ? (stream_id % 4 == 0) : (stream_id % 4 == 1);
		@@ -5479,6 +5517,36 @@ void quic::resetStream(uint64_t stream_id, uint64_t error_code) {
		if (it != _streams.end()) {
		it->second.send_fin = true;
		}

		retireStreamIfDone(stream_id);
		}

		// ============================================================================
		// STOP_SENDING: ask the peer to abort its send side (RFC 9000 §19.5)
		// ============================================================================

		void quic::stopSending(uint64_t stream_id, uint64_t error_code) {
		std::lock_guard<std::recursive_mutex> lock(quic_mtx());

		if (_streams.find(stream_id) == _streams.end()) return;

		std::vector<uint8_t> frame;
		frame.push_back(0x05); // STOP_SENDING frame type
		uint8_t buf[8];
		size_t bytes = encodeVarInt(stream_id, buf);
		frame.insert(frame.end(), buf, buf + bytes);
		bytes = encodeVarInt(error_code, buf);
		frame.insert(frame.end(), buf, buf + bytes);

		std::vector<uint8_t> packet = buildShortHeaderPacket(frame);
		const uint8_t* hp_key = _is_server ? _app_hp_server : _app_hp_client;
		applyHeaderProtection(packet, hp_key);
		ssize_t sent = sendPacket(packet.data(), packet.size());
		if (sent < 0) {
		// Best-effort immediate retry — do not sleep here since the
		// caller typically holds quic_mtx().
		sendPacket(packet.data(), packet.size());
		}
		}

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

src/socket.h

+17 −0

Original line number	Diff line number	Diff line
		@@ -650,6 +650,14 @@ namespace netplus {
		// transfer instead of silently accepting a truncated body as complete.
		void resetStream(uint64_t stream_id, uint64_t error_code = 0x0102);

		// Tell the peer to stop sending on a stream whose response we no
		// longer want (RFC 9000 §19.5 STOP_SENDING) — e.g. abandoning a
		// redundant/slow fetch once enough other responses arrived. Without
		// this, the peer has no signal that we stopped listening, so it may
		// not finish its send side for a long time (or ever), and neither
		// side ever retires the stream or reclaims its stream-count credit.
		void stopSending(uint64_t stream_id, uint64_t error_code = 0x0102);

		// Read raw UDP datagrams and process QUIC packets without consuming
		// stream data. Use this in HTTP/3 client loops to pump the network
		// while reading specific streams via recvStreamData().
		@@ -702,6 +710,15 @@ namespace netplus {
		void processStreamFrame(const uint8_t* data, size_t len, size_t& offset);
		void processAckFrame(const uint8_t* data, size_t len, size_t& offset);

		// If a stream has terminated in both directions (FIN/RESET on both
		// send and recv sides), erase it and — for a peer-initiated stream —
		// grant the peer fresh stream-count credit via MAX_STREAMS. Called
		// from every place a stream's send_fin/recv_fin can flip to true
		// (closeStream, resetStream, and STOP_SENDING/RESET_STREAM/STREAM
		// frame processing) so retirement fires as soon as both sides agree
		// the stream is done, regardless of which side notices last.
		void retireStreamIfDone(uint64_t stream_id);

		// Loss detection and retransmission
		void checkLossAndRetransmit();
		void recordSentPacket(uint64_t pn, uint64_t stream_id, uint64_t stream_offset,