Commit 65fc51fa authored by jan.koester's avatar jan.koester
Browse files

quic: reap idle and draining server children, close on destruction 

The listener's registry purge only removed Closed children, so every
abandoned connection stayed in memory forever (~200 kB each):
- a received CONNECTION_CLOSE leaves the child Draining, never Closed
- the idle timeout was only negotiated, never enforced; our own
  keepalive PINGs even refreshed _last_activity of a vanished peer
- ~quic() sent no CONNECTION_CLOSE, so a client dropping a connection
  (paritypp's socket.reset()) never told the server

Track the last successfully processed frame in _last_recv_ticks and let
the purge also drop children that finished draining (3s) or stayed
silent past _max_idle_timeout. ~quic() now closes non-child connections.

Seen as steadily growing RSS of fenrirds and mediadb in cluster mode.

Co-Authored-By: default avatarClaude Opus 5.5 <noreply@anthropic.com>
parent aa44443a

src/quic.cpp

+43 −1
Changes for src/quic.cpp: 43 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -494,6 +494,18 @@ void quic::getSocketBufferSizes(size_t& rcvbuf_bytes, size_t& sndbuf_bytes) cons 
quic::~quic() {

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



    // A client connection dropped without close() (paritypp's

    // socket.reset() on reconnect/failure) used to vanish silently,

    // leaving the server's child for it alive until its idle timeout.

    // Tell the peer. Children are only destroyed once Closed (registry

    // purge), so close() would be a no-op for them anyway.

    if (_parent == nullptr) {

        try {

            close();

        } catch (...) {

        }

    }



    // Deliberately does NOT touch _parent->_child_connections: this

    // destructor can run as a direct side effect of the registry itself

    // dropping the last shared_ptr to this connection (accept()'s purge
@@ -2513,7 +2525,7 @@ void quic::accept(std::unique_ptr<socket>& csock, bool nonblock) { 
            std::vector<std::shared_ptr<quic>> to_destroy;

            std::unique_lock<netplus::WriterPreferringSharedMutex> reg_lock(_registry_mutex);

            for (auto it = _child_connections.begin(); it != _child_connections.end(); ) {

                if (it->second && it->second->_conn_state.load() == ConnectionState::Closed) {

                if (it->second && it->second->expiredForPurge(purge_now)) {

                    to_destroy.push_back(std::move(it->second));

                    it = _child_connections.erase(it);

                } else {
@@ -3791,6 +3803,31 @@ void quic::close() { 
    // entry in place briefly is harmless.

}



// Before this, only Closed children were ever purged: a peer's

// CONNECTION_CLOSE (or our own protocol-error close) left a child in

// Draining forever, and a peer that just disappeared left it Connected

// forever — the listener (paritypp::server in fenrirds/mediadb) grew by

// one full connection per abandoned client connection.

bool quic::expiredForPurge(std::chrono::steady_clock::time_point now) {

    ConnectionState st = _conn_state.load();

    if (st == ConnectionState::Closed) return true;



    auto last_recv = std::chrono::steady_clock::time_point(

        std::chrono::steady_clock::duration(_last_recv_ticks.load(std::memory_order_relaxed)));

    auto silent = now - last_recv;



    if (st == ConnectionState::Draining) {

        if (silent < DRAINING_PERIOD) return false;

    } else {

        // Idle timeout closes silently (RFC 9000 §10.1). _max_idle_timeout

        // already holds the negotiated min of both sides' values.

        if (_max_idle_timeout == 0 ||

            silent < std::chrono::milliseconds(_max_idle_timeout)) return false;

    }

    _conn_state.store(ConnectionState::Closed);

    return true;

}



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

// Frame Building: CONNECTION_CLOSE

// ============================================================================
@@ -5043,6 +5080,11 @@ void quic::abortFrameParsing(size_t len, size_t& offset, uint64_t error_code, co 
void quic::processFrame(const uint8_t* data, size_t len, size_t& offset, bool& ack_eliciting) {

    if (offset >= len) return;



    // Only reached for packets that decrypted successfully — spoofed

    // datagrams can't keep a connection alive past its idle timeout.

    _last_recv_ticks.store(std::chrono::steady_clock::now().time_since_epoch().count(),

                           std::memory_order_relaxed);



    // Read frame type

    size_t type_bytes;

    uint64_t frame_type = decodeVarInt(&data[offset], len - offset, type_bytes);

src/socket.h

+15 −0
Changes for src/socket.h: 15 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -2083,6 +2083,21 @@ namespace netplus { 
		std::chrono::steady_clock::time_point _last_activity;

		static constexpr int KEEPALIVE_INTERVAL_MS = 15000;  // Send PING every 15s of inactivity



		// Last time a frame from the peer was successfully decrypted and

		// processed (RFC 9000 §10.1 idle timer). Unlike _last_activity it

		// is NOT refreshed by our own sends — our keepalive PINGs to a

		// vanished peer would otherwise keep a server child alive forever.

		// Atomic: written by whichever thread processes this connection's

		// packets, read by the listener's registry purge in accept().

		std::atomic<std::chrono::steady_clock::rep> _last_recv_ticks{

			std::chrono::steady_clock::now().time_since_epoch().count()};

		// Draining period after a CONNECTION_CLOSE (RFC 9000 §10.2 asks

		// for 3×PTO; a fixed bound is enough to swallow stragglers).

		static constexpr std::chrono::milliseconds DRAINING_PERIOD{3000};

		// accept()'s registry purge: true if this child is Closed, done

		// draining, or idle past _max_idle_timeout (marks it Closed).

		bool expiredForPurge(std::chrono::steady_clock::time_point now);



		// Handshake retransmission: store last handshake flight for retransmit

		std::vector<std::vector<uint8_t>> _hs_flight_packets;  // Stored for retransmit

		std::chrono::steady_clock::time_point _hs_last_send;