daemon/quic_conn: fix max_stream extention, use libuv times for idle timeouts

closing streams doesn't increase the number of available streams automatically, for connections with streams over the initial limit this caused connections to idle until closed

diff --git a/daemon/quic_conn.c b/daemon/quic_conn.c
index 3d2c325beae779bc867f09789c24dd342087086b..dd196fad97fc351594f1f14560ab2ee764ea6615 100644 (file)
--- a/daemon/quic_conn.c
+++ b/daemon/quic_conn.c
@@ -3,6 +3,7 @@
  */
 
 #include "quic_conn.h"
+#include "lib/log.h"
 #include "quic_stream.h"
 #include "quic_common.h"
 #include "libdnssec/random.h"
@@ -11,6 +12,7 @@
 #include "worker.h"
 #include <libknot/wire.h>
 #include <ngtcp2/ngtcp2.h>
+#include <uv.h>
 
 #define EPHEMERAL_CERT_EXPIRATION_SECONDS_RENEW_BEFORE ((time_t)60*60*24*7)
 
@@ -95,8 +97,6 @@ static int kr_recv_stream_data_cb(ngtcp2_conn *ngconn, uint32_t flags,
        int64_t stream_id, uint64_t offset, const uint8_t *data,
        size_t datalen, void *user_data, void *stream_user_data)
 {
-       (void)ngconn;
-
        struct pl_quic_conn_sess_data *conn = user_data;
        struct pl_quic_stream_sess_data *stream = stream_user_data;
 
@@ -135,6 +135,11 @@ static int kr_recv_stream_data_cb(ngtcp2_conn *ngconn, uint32_t flags,
                kr_require(wire_buf_consume(&stream->pers_inbuf, datalen - sizeof(uint16_t)) == kr_ok());
        }
 
+       /* we can ignore ret return value, it can only be ENOMEM, at which point
+        * there is nothing we can do anyway and the connection will timeout cleanly*/
+       (void)ngtcp2_conn_extend_max_stream_offset(ngconn, stream_id, datalen);
+       ngtcp2_conn_extend_max_offset(ngconn, datalen);
+
 finished:
        if (flags & NGTCP2_STREAM_DATA_FLAG_FIN) {
                queue_push(conn->pending_unwrap, stream);
@@ -181,7 +186,10 @@ static int stream_open_cb(ngtcp2_conn *ngconn,
                                1,
                                false);
 
-       kr_require(new_subsession);
+       if (!new_subsession) {
+               kr_log_error(DOQ, "Failed to create new quic stream session\n");
+               return kr_error(ENOMEM);
+       }
 
        struct pl_quic_stream_sess_data *stream =
                protolayer_sess_data_get_proto(new_subsession,
@@ -189,7 +197,6 @@ static int stream_open_cb(ngtcp2_conn *ngconn,
        kr_require(stream);
 
        stream->conn_ref = conn;
-
        if (conn->streams_count <= 0) {
                add_head(&conn->streams, &stream->list_node);
        } else {
@@ -207,8 +214,7 @@ static int stream_close_cb(ngtcp2_conn *ngconn, uint32_t flags,
                int64_t stream_id, uint64_t app_error_code,
                void *user_data, void *stream_user_data)
 {
-       (void)ngconn;
-
+       ngtcp2_conn_extend_max_streams_bidi(ngconn, 1);
        struct pl_quic_conn_sess_data *conn = user_data;
        struct pl_quic_stream_sess_data *stream = stream_user_data;
        rem_node(&stream->list_node);
@@ -231,8 +237,7 @@ static int get_new_connection_id_cb(ngtcp2_conn *ngconn, ngtcp2_cid *cid,
        (void)ngconn;
        struct pl_quic_conn_sess_data *conn = user_data;
        session2_event(conn->h.session->transport.parent,
-                       PROTOLAYER_EVENT_CONNECT_UPDATE,
-                       conn);
+                       PROTOLAYER_EVENT_CONNECT_UPDATE, conn);
        memcpy(cid, &conn->dcid, sizeof(ngtcp2_cid));
 
        if (ngtcp2_crypto_generate_stateless_reset_token(token, conn->secret,
@@ -595,11 +600,6 @@ int send_special(struct pl_quic_conn_sess_data *conn,
        struct wire_buf *save = ctx->payload.wire_buf;
        ctx->payload.wire_buf = &err_wb;
 
-       // if (wire_buf_data_length(ctx->payload.wire_buf) != 0) {
-       //      wire_buf_reset(ctx->payload.wire_buf);
-       //      return kr_error(EINVAL);
-       // }
-
        uint64_t now = quic_timestamp();
 
        ngtcp2_cid new_dcid;
@@ -673,7 +673,6 @@ int send_special(struct pl_quic_conn_sess_data *conn,
                } else {
                        ccerr.type = NGTCP2_CCERR_TYPE_APPLICATION;
                }
-               // ccerr.error_code = NGTCP2_ERR_HANDSHAKE_TIMEOUT;
                ret = ngtcp2_conn_write_connection_close(
                        conn->conn, NULL, &pi,
                        wire_buf_free_space(ctx->payload.wire_buf),
@@ -743,6 +742,7 @@ static enum protolayer_iter_cb_result pl_quic_conn_unwrap(void *sess_data,
                }
        }
 
+       uv_timer_again(&conn->h.session->timer);
        ret = handle_packet(conn, ctx);
        if (ret != kr_ok()) {
                if (QUIC_CAN_SEND(conn)) {
@@ -888,6 +888,7 @@ static int pl_quic_conn_sess_init(struct session2 *session, void *sess_data, voi
 
        conn->conn = NULL;
        conn->priority = NULL;
+       conn->streams_count = 0;
        conn->tls_session = NULL;
        conn->server_credentials = NULL;
        if (quic_generate_secret(conn->secret, sizeof(conn->secret)) != kr_ok()) {
@@ -896,6 +897,10 @@ static int pl_quic_conn_sess_init(struct session2 *session, void *sess_data, voi
                return kr_error(EINVAL);
        }
 
+       session2_timer_start(session, PROTOLAYER_EVENT_CONNECT_TIMEOUT,
+                       QUIC_CONN_IDLE_TIMEOUT / NGTCP2_MILLISECONDS,
+                       QUIC_CONN_IDLE_TIMEOUT / NGTCP2_MILLISECONDS);
+
        return kr_ok();
 }
 
@@ -919,7 +924,6 @@ static int pl_quic_conn_sess_deinit(struct session2 *session, void *sess_data)
                kr_log_error(DOQ, "Client side of QUIC is not implemented\n");
        }
 
-
        conn->priority = NULL;
        conn->tls_session = NULL;
        conn->server_credentials = NULL;
@@ -935,6 +939,11 @@ static enum protolayer_event_cb_result pl_quic_conn_event_unwrap(
                struct session2 *session, void *sess_data)
 {
        struct pl_quic_conn_sess_data *conn = sess_data;
+       if (event == PROTOLAYER_EVENT_CONNECT_TIMEOUT) {
+               session2_event(conn->h.session->transport.parent, event, conn);
+               return PROTOLAYER_EVENT_CONSUME;
+       }
+
        if (event == PROTOLAYER_EVENT_DISCONNECT ||
                        event == PROTOLAYER_EVENT_CLOSE ||
                        event == PROTOLAYER_EVENT_FORCE_CLOSE) {
@@ -948,6 +957,7 @@ static enum protolayer_event_cb_result pl_quic_conn_event_unwrap(
                        --conn->streams_count;
                }
                session2_dec_refs(session);
+               session2_timer_stop(session);
                return PROTOLAYER_EVENT_CONSUME;
        }
 

diff --git a/daemon/quic_demux.c b/daemon/quic_demux.c
index 72a6e52da699807511c82844c8125ddd3f660ae2..6ba238826f18258632b3c8176846d75dfa8829de 100644 (file)
--- a/daemon/quic_demux.c
+++ b/daemon/quic_demux.c
@@ -129,6 +129,7 @@ static void send_excessive_load(struct pl_quic_conn_sess_data *conn,
        (void)send_special(conn, ctx, DOQ_EXCESSIVE_LOAD);
 }
 
+/* unused for now, compare performance with per conn uv_timer_t spawns */
 void kr_quic_table_sweep(struct kr_quic_table *table,
                struct protolayer_iter_ctx *ctx)
 {
@@ -245,20 +246,23 @@ static enum protolayer_iter_cb_result pl_quic_demux_unwrap(void *sess_data,
        qconn = kr_quic_table_lookup(&dcid, demux->conn_table);
        if (!qconn) {
                /* Clear idle connections */
-               kr_quic_table_sweep(demux->conn_table, ctx);
+               // kr_quic_table_sweep(demux->conn_table, ctx);
 
                if (demux->conn_table->usage >= demux->conn_table->max_conns) {
                        kr_log_warning(DOQ,
                                "Refusing to open new connection, reached limit of active conns\n");
-                       /* we may inform the client that limits have been reached */
+                       /* we might want to inform the client
+                        * that limits have been reached */
                        return protolayer_break(ctx, kr_ok());
                }
 
                ngtcp2_pkt_hd header = { 0 };
-               if (ngtcp2_accept(&header,
+               ret = ngtcp2_accept(&header,
                        wire_buf_data(ctx->payload.wire_buf),
-                       wire_buf_data_length(ctx->payload.wire_buf))
-                               != NGTCP2_NO_ERROR) {
+                       wire_buf_data_length(ctx->payload.wire_buf));
+               if (ret != NGTCP2_NO_ERROR) {
+                       kr_log_debug(DOQ, "error accepting new conn: %s (%d)\n",
+                                       ngtcp2_strerror(ret), ret);
 
                        /* either the packet is not acceptable as the first
                         * packet of a new connection, or the function failed
@@ -332,7 +336,6 @@ static enum protolayer_iter_cb_result pl_quic_demux_unwrap(void *sess_data,
                                        PROTOLAYER_TYPE_QUIC_CONN);
                kr_quic_table_add(conn_sess_data, &dcid,
                                demux->conn_table);
-
                qconn = conn_sess_data;
        }
 
@@ -343,7 +346,7 @@ static enum protolayer_iter_cb_result pl_quic_demux_unwrap(void *sess_data,
                        ctx->finished_cb_baton);
 
        quic_conn_mark_used(qconn, demux->conn_table);
-       kr_quic_table_sweep(demux->conn_table, ctx);
+       // kr_quic_table_sweep(demux->conn_table, ctx);
 
        return protolayer_break(ctx, kr_ok());
 }
@@ -512,7 +515,6 @@ static int pl_quic_demux_sess_deinit(struct session2 *session, void *data)
 {
        struct pl_quic_demux_sess_data *quic = data;
        kr_quic_table_free(quic->conn_table);
-
        return kr_ok();
 }
 
@@ -589,7 +591,8 @@ static enum protolayer_event_cb_result pl_quic_demux_event_unwrap(
                return PROTOLAYER_EVENT_CONSUME;
        }
 
-       if (event == PROTOLAYER_EVENT_DISCONNECT) {
+       if (event == PROTOLAYER_EVENT_DISCONNECT ||
+                       event == PROTOLAYER_EVENT_CONNECT_TIMEOUT) {
                if (*baton == NULL)
                        return PROTOLAYER_EVENT_CONSUME;
 

diff --git a/daemon/quic_stream.c b/daemon/quic_stream.c
index b9f33df57bc444099aa24e2899fce72751588c2d..9575be899ddaff7182558704e6d5301cc615283f 100644 (file)
--- a/daemon/quic_stream.c
+++ b/daemon/quic_stream.c
@@ -190,7 +190,6 @@ void kr_quic_stream_ack_data(struct pl_quic_stream_sess_data *stream,
 {
        kr_require(stream);
        struct list *obs = &stream->outbufs;
-
        struct kr_quic_obuf *first;
 
        while (!EMPTY_LIST(*obs) && end_acked >=
@@ -231,6 +230,7 @@ int update_stream_pers_buffer(struct pl_quic_stream_sess_data *stream,
 static int pl_quic_stream_sess_deinit(struct session2 *session, void *sess_data)
 {
        struct pl_quic_stream_sess_data *stream = sess_data;
+       ngtcp2_conn_shutdown_stream(stream->conn, 0, stream->stream_id, 0);
        kr_require(queue_len(session->waiting) <= 0);
        kr_quic_stream_ack_data(stream, stream->stream_id, SIZE_MAX, false);
        wire_buf_deinit(&stream->pers_inbuf);