testt

                batch_count = 0;

            }

            // Drain incoming to receive ACKs and free cwnd

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

            // First: spin-pump a couple of times without waiting (ACK may

            // already be sitting in the socket buffer, arrived in the

            // microseconds since the last check). Capped at 3, not 8:

            // measured under sustained cwnd-saturated bulk transfer, this

            // resolves in ~2.1 iterations on average — a cap of 8 almost

            // never gets used, it just means up to 5 extra unconditional

            // recvmmsg() calls per stall (at 50k+ stalls/sec measured, a

            // real syscall cost) on the rare tail that doesn't resolve

            // quickly. Genuinely slow-to-resolve stalls fall through to the

            // wait loop below either way.

            bool unblocked = false;

            int spin_count = 0;

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

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

                // 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).

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

                pumpIncomingLocked();

                _flushes_since_pump = 0;

                ++spin_count;

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

            }

            quicPerf().cc_spin_iterations.fetch_add(static_cast<uint64_t>(spin_count), std::memory_order_relaxed);

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

            // (and its epoll fd) across every retry instead of constructing

            // a fresh one per ~1ms attempt — see _cc_socketwait's comment.

            // If still blocked, wait with a real timeout instead of

            // spinning further. Reuse one socketwait (and its epoll fd)

            // across every retry instead of constructing a fresh one each

            // time — see _cc_socketwait's comment.

            if (!unblocked) {

                if (!_cc_socketwait) _cc_socketwait = std::make_unique<socketwait>();

                // Budget the wait off the connection's own RTO estimate

                // baseline). Floor at 100ms to match prior behavior on

                // healthy connections; cap at 2s so a blown-up RTO

                // estimate can't stall the sender indefinitely.

                int cc_wait_budget_ms = static_cast<int>(

                    std::clamp(_rto * 1000.0, 100.0, 2000.0));

                double cc_wait_budget_ms = std::clamp(_rto * 1000.0, 100.0, 2000.0);

                auto wait_deadline = std::chrono::steady_clock::now() +

                    std::chrono::duration_cast<std::chrono::steady_clock::duration>(

                        std::chrono::duration<double, std::milli>(cc_wait_budget_ms));

                // Each waitRead() call already returns the instant the

                // socket becomes readable — epoll_wait() doesn't sleep for

                // the full requested timeout just because a large value was

                // asked for, it wakes up on the first read-ready event.

                // There is therefore no responsiveness reason to slice a

                // multi-hundred-millisecond budget into a poll loop with a

                // fixed tiny per-call timeout: that pattern only forces a

                // wasted epoll_ctl(ADD)+epoll_wait+epoll_ctl(DEL) cycle

                // (and a follow-up recvmmsg() that finds almost nothing —

                // exactly the "recvmmsg used like a per-packet poller"

                // anti-pattern) on every tick where nothing happened, one

                // real syscall per elapsed millisecond of true idle time.

                // Instead, each iteration waits for however much of the

                // budget remains, only capped (kWaitSliceCapMs) so a

                // connection that transitions to Closed/Draining mid-wait

                // is still noticed in bounded time rather than only after

                // the entire (up to 2s) budget elapses.

                static constexpr int kWaitSliceCapMs = 20;

                int wait_count = 0;

                for (int cc_wait = 0; cc_wait < cc_wait_budget_ms; ++cc_wait) {

                while (true) {

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

                        _conn_state.load() == ConnectionState::Draining) break;

                    auto now = std::chrono::steady_clock::now();

                    if (now >= wait_deadline) break;

                    int remaining_ms = static_cast<int>(std::chrono::duration_cast<

                        std::chrono::milliseconds>(wait_deadline - now).count());

                    if (remaining_ms <= 0) break;

                    int this_wait_ms = std::min(remaining_ms, kWaitSliceCapMs);

                    lock.unlock();

                    _cc_socketwait->waitRead(*this, 1);

                    _cc_socketwait->waitRead(*this, this_wait_ms);

                    lock.lock();

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

                    // comment on pumpIncomingLocked() (F31).