test

            // First: spin-pump a few times without syscall (ACK may already be queued)

            bool unblocked = false;

            for (int spin = 0; spin < 8; ++spin) {

                pumpIncoming();

                // quic_mtx() is already held throughout this function (see

                // `lock` above) — pumpIncomingLocked() skips the redundant

                // recursive re-lock pumpIncoming() would otherwise do here,

                // up to 8 times per congestion-window stall (F31).

                pumpIncomingLocked();

                if (cwndAllowsSend(est_pkt_size)) { unblocked = true; break; }

            }

            // If still blocked, poll with timeout. Reuse one socketwait

                    lock.unlock();

                    _cc_socketwait->waitRead(*this, 1);

                    lock.lock();

                    pumpIncoming();

                    // Re-locked just above — see the spin-loop's identical

                    // comment on pumpIncomingLocked() (F31).

                    pumpIncomingLocked();

                    if (cwndAllowsSend(est_pkt_size)) { unblocked = true; break; }

                }

            }

            flushBatch();

            batch_count = 0;

            // Pace: pump incoming every batch to process ACKs/FC updates

            pumpIncoming();

            // Pace: pump incoming every batch to process ACKs/FC updates.

            // quic_mtx() is already held throughout this function (F31).

            pumpIncomingLocked();

            if (_conn_state.load() == ConnectionState::Closed ||

                _conn_state.load() == ConnectionState::Draining) {

                break;

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

void quic::pumpIncoming() {

    std::lock_guard<std::recursive_mutex> lock(quic_mtx());

    pumpIncomingLocked();

}

void quic::pumpIncomingLocked() {

    quic* parent = _parent ? _parent : this;

    // Use udp::recvBatch (sendmmsg/recvmmsg + GRO) to read datagrams. Safe

    }

    for (auto& [t, pkts] : grouped) {

        if (t == this) {

            // pumpIncomingLocked()'s precondition is that quic_mtx() on

            // `this` is already held by the caller — re-locking it here

            // (as the general t != this branch below does for every other

            // connection's own, genuinely different mutex) would just be

            // two redundant recursive lock/unlock cycles on top of the

            // caller's, wasted every time this runs from

            // sendStreamData()'s congestion-window-wait spin loop (F31).

            // The two sub-blocks below are unreachable for t == this

            // anyway (both explicitly guarded on t != this), so there's

            // nothing else this branch needs to do.

            t->processApplicationPacketsBatch(pkts);

            continue;

        }

        std::unique_lock<std::recursive_mutex> target_lock(t->quic_mtx());

        t->processApplicationPacketsBatch(pkts);

        // The self case's pending flow-control frames are flushed by the

    }

    // Flush own pending flow control frames, and check for lost packets to

    // retransmit — both operate on `this`'s own state.

    std::lock_guard<std::recursive_mutex> lock(quic_mtx());

    // retransmit — both operate on `this`'s own already-locked state (see

    // pumpIncomingLocked()'s precondition).

    flushPendingFlowControlAndCheckLoss();

}

void quic::flushPendingFlowControlAndCheckLoss() {

    if (_handshake_complete && _conn_state.load() == ConnectionState::Connected) {

        for (auto& [sid, limit] : _pending_max_stream_data) {

            sendMaxStreamData(sid, limit);

		void pumpIncoming();

	private:

		// Body of pumpIncoming(), assuming quic_mtx() on `this` is already

		// held by the caller (recursive_mutex, so re-entering it isn't

		// incorrect — just two extra uncontended lock/unlock cycles per

		// call, for the self-targeted case only, that add up when this

		// runs repeatedly from sendStreamData()'s congestion-window-wait

		// spin loop). Every other connection's own mutex (`t->quic_mtx()`

		// for `t != this` in the grouped-datagram loop) is still acquired

		// normally — those are genuinely different locks, not redundant.

		void pumpIncomingLocked();

		// Shared tail of pumpIncoming()/pumpIncomingLocked(): flush pending

		// flow-control frames and check for retransmits, both operating on

		// `this`'s own already-locked state.

		void flushPendingFlowControlAndCheckLoss();

		// Current encryption level for frame processing. Declared here

		// (rather than down with the other crypto state) because it's a

		// parameter type for protectPacket()/unprotectPacket() below, and a