[haproxy.git] /

#include <haproxy/mux_quic.h>
#include <haproxy/mux_quic_priv.h>

#include <import/eb64tree.h>

#include <haproxy/api.h>
#include <haproxy/buf.h>
#include <haproxy/chunk.h>
#include <haproxy/connection.h>
#include <haproxy/dynbuf.h>
#include <haproxy/global-t.h>
#include <haproxy/h3.h>
#include <haproxy/list.h>
#include <haproxy/ncbuf.h>
#include <haproxy/pool.h>
#include <haproxy/proxy.h>
#include <haproxy/qcm_http.h>
#include <haproxy/qcm_qmux.h>
#include <haproxy/qcm_trace.h>
#include <haproxy/quic_conn.h>
#include <haproxy/quic_enc.h>
#include <haproxy/quic_fctl.h>
#include <haproxy/quic_frame.h>
#include <haproxy/quic_pacing.h>
#include <haproxy/quic_sock.h>
#include <haproxy/quic_stream.h>
#include <haproxy/quic_stats.h>
#include <haproxy/quic_tp-t.h>
#include <haproxy/quic_tune.h>
#include <haproxy/quic_tx.h>
#include <haproxy/quic_utils.h>
#include <haproxy/session.h>
#include <haproxy/ssl_sock-t.h>
#include <haproxy/stconn.h>
#include <haproxy/time.h>
#include <haproxy/trace.h>
#include <haproxy/xprt_qmux.h>
#include <haproxy/xref.h>

DECLARE_TYPED_POOL(pool_head_qcc, "qcc", struct qcc);
DECLARE_TYPED_POOL(pool_head_qcs, "qcs", struct qcs);
DECLARE_STATIC_TYPED_POOL(pool_head_qc_stream_rxbuf, "qc_stream_rxbuf", struct qc_stream_rxbuf);

static void qcm_ctrl_send(struct qc_stream_desc *, uint64_t data, uint64_t offset);
static void qcm_ctrl_room(struct qc_stream_desc *, uint64_t room);

static void qcc_release(struct qcc *qcc);
static int qcc_app_init(struct qcc *qcc);
static void qcc_app_shutdown(struct qcc *qcc);

/* Returns true if pacing should be used for <conn> connection. */
static int qcc_is_pacing_active(const struct connection *conn)
{
        return conn_is_quic(conn) && quic_tune_conn_test(QUIC_TUNE_FB_TX_PACING, conn);
}

/* Free <rxbuf> instance and its inner data storage attached to <qcs> stream. */
static void qcs_free_rxbuf(struct qcs *qcs, struct qc_stream_rxbuf *rxbuf)
{
        struct ncbuf *ncbuf;
        struct buffer buf;

        ncbuf = &rxbuf->ncb;
        if (!ncb_is_null(ncbuf)) {
                buf = b_make(ncbuf->area, ncbuf->size, 0, 0);
                b_free(&buf);
                offer_buffers(NULL, 1);
        }
        rxbuf->ncb = NCBUF_NULL;

        /* Reset DEM_FULL as buffer is released. This ensures mux is not woken
         * up from rcv_buf stream callback when demux was previously blocked.
         */
        qcs->flags &= ~QC_SF_DEM_FULL;

        eb64_delete(&rxbuf->off_node);
        pool_free(pool_head_qc_stream_rxbuf, rxbuf);
        bdata_ctr_bdec(&qcs->rx.data);
}

/* Free <qcs> instance. This function is reserved for internal usage : it must
 * only be called on qcs alloc error or on connection shutdown. Else
 * qcs_destroy must be preferred to handle QUIC flow-control increase.
 */
static void qcs_free(struct qcs *qcs)
{
        struct qcc *qcc = qcs->qcc;
        struct qc_stream_rxbuf *b;

        TRACE_ENTER(QMUX_EV_QCS_END, qcc->conn, qcs);
        TRACE_STATE("releasing QUIC stream", QMUX_EV_QCS_END, qcc->conn, qcs);

        /* Safe to use even if already removed from the list. */
        LIST_DEL_INIT(&qcs->el_opening);
        LIST_DEL_INIT(&qcs->el_recv);
        LIST_DEL_INIT(&qcs->el_send);
        LIST_DEL_INIT(&qcs->el_fctl);
        LIST_DEL_INIT(&qcs->el_buf);

        /* Release stream endpoint descriptor. */
        BUG_ON(qcs->sd && !se_fl_test(qcs->sd, SE_FL_ORPHAN));
        sedesc_free(qcs->sd);
        qcs->sd = NULL;

        if (qcs->flags & QC_SF_HREQ_RECV) {
                BUG_ON(!qcc->nb_hreq);
                --qcc->nb_hreq;
        }

        /* Release app-layer context. */
        if (qcs->ctx && qcc->app_ops->detach)
                qcc->app_ops->detach(qcs);

        /* Release qc_stream_desc buffer from quic-conn layer. */
        if (conn_is_quic(qcc->conn) && qcs->tx.stream) {
                qc_stream_desc_sub_send(qcs->tx.stream, NULL);
                qc_stream_desc_release(qcs->tx.stream, qcs->tx.fc.off_real, qcc);
        }
        else if (!conn_is_quic(qcc->conn)) {
                b_free(&qcs->tx.qmux_buf);
        }

        /* Free Rx buffer. */
        while (!eb_is_empty(&qcs->rx.bufs)) {
                b = container_of(eb64_first(&qcs->rx.bufs),
                                 struct qc_stream_rxbuf, off_node);
                qcs_free_rxbuf(qcs, b);
        }

        if (b_size(&qcs->rx.app_buf)) {
                b_free(&qcs->rx.app_buf);
                offer_buffers(NULL, 1);
        }

        /* Remove qcs from qcc tree. */
        eb64_delete(&qcs->by_id);

        pool_free(pool_head_qcs, qcs);

        TRACE_LEAVE(QMUX_EV_QCS_END, qcc->conn);
}

/* Allocate a new QUIC streams with id <id> and type <type>. */
static struct qcs *qcs_new(struct qcc *qcc, uint64_t id, enum qcs_type type)
{
        struct qcs *qcs;

        TRACE_ENTER(QMUX_EV_QCS_NEW, qcc->conn);

        qcs = pool_alloc(pool_head_qcs);
        if (!qcs) {
                TRACE_ERROR("alloc failure", QMUX_EV_QCS_NEW, qcc->conn);
                return NULL;
        }

        qcs->tx.stream = NULL;
        qcs->qcc = qcc;
        qcs->sess = NULL;
        qcs->sd = NULL;
        qcs->flags = QC_SF_NONE;
        qcs->st = QC_SS_IDLE;
        qcs->ctx = NULL;

        /* App callback attach may register the stream for http-request wait.
         * These fields must be initialized before.
         */
        LIST_INIT(&qcs->el_opening);
        LIST_INIT(&qcs->el_recv);
        LIST_INIT(&qcs->el_send);
        LIST_INIT(&qcs->el_fctl);
        LIST_INIT(&qcs->el_buf);
        qcs->start = TICK_ETERNITY;

        /* store transport layer stream descriptor in qcc tree */
        qcs->id = qcs->by_id.key = id;
        eb64_insert(&qcc->streams_by_id, &qcs->by_id);

        /* Different limits can be set by the peer for local and remote bidi streams. */
        if (quic_stream_is_bidi(id)) {
                qfctl_init(&qcs->tx.fc, quic_stream_is_local(qcc, id) ?
                                        qcc->rfctl.msd_bidi_r : qcc->rfctl.msd_bidi_l);
        }
        else if (quic_stream_is_local(qcc, id)) {
                qfctl_init(&qcs->tx.fc, qcc->rfctl.msd_uni_l);
        }
        else {
                qfctl_init(&qcs->tx.fc, 0);
        }

        qcs->tx.msd_frm = NULL;

        qcs->rx.bufs = EB_ROOT_UNIQUE;
        qcs->rx.app_buf = BUF_NULL;
        qcs->rx.offset = qcs->rx.offset_max = 0;

        if (quic_stream_is_bidi(id)) {
                qcs->rx.msd = quic_stream_is_local(qcc, id) ? qcc->lfctl.msd_bidi_l :
                                                              qcc->lfctl.msd_bidi_r;
        }
        else if (quic_stream_is_remote(qcc, id)) {
                qcs->rx.msd = qcc->lfctl.msd_uni_r;
        }
        qcs->rx.msd_base = 0;
        bdata_ctr_init(&qcs->rx.data);

        qcs->wait_event.tasklet = NULL;
        qcs->wait_event.events = 0;
        qcs->subs = NULL;

        qcs->rs_err = qcs->ss_err = 0;

        /* Reset all timers and start base one. */
        tot_time_reset(&qcs->timer.base);
        tot_time_reset(&qcs->timer.buf);
        tot_time_reset(&qcs->timer.fctl);
        tot_time_start(&qcs->timer.base);

        /* Allocate transport layer stream descriptor. Only needed for TX. */
        if (!quic_stream_is_uni(id) || !quic_stream_is_remote(qcc, id)) {
                if (conn_is_quic(qcc->conn)) {
                        struct quic_conn *qc = qcc->conn->handle.qc;
                        qcs->tx.stream = qc_stream_desc_new(id, type, qcs, qc);
                        if (!qcs->tx.stream) {
                                TRACE_ERROR("qc_stream_desc alloc failure", QMUX_EV_QCS_NEW, qcc->conn, qcs);
                                goto err;
                        }

                        qc_stream_desc_sub_send(qcs->tx.stream, qcm_ctrl_send);
                        qc_stream_desc_sub_room(qcs->tx.stream, qcm_ctrl_room);
                }
                else {
                        qcs->tx.qmux_buf = BUF_NULL;
                }
        }

        /* If app layer is closed, reset immediately a new stream if proto does
         * not implement graceful shutdown. Else proto is responsible to either
         * accept or reject the new stream via its attach() operation.
         */
        if (qcc->app_st >= QCC_APP_ST_SHUT && !qcc->app_ops->shutdown) {
                qcc_abort_stream_read(qcs, 0);
                qcc_reset_stream(qcs, 0, 0);
                goto out;
        }
        else if (qcc->app_ops->attach && qcc->app_ops->attach(qcs, qcc->ctx)) {
                TRACE_ERROR("app proto failure", QMUX_EV_QCS_NEW, qcc->conn, qcs);
                goto err;
        }

 out:
        TRACE_STATE("created new QUIC stream", QMUX_EV_QCS_NEW, qcc->conn, qcs);
        TRACE_LEAVE(QMUX_EV_QCS_NEW, qcc->conn, qcs);
        return qcs;

 err:
        qcs_free(qcs);
        TRACE_LEAVE(QMUX_EV_QCS_NEW, qcc->conn);
        return NULL;
}

static forceinline struct stconn *qcs_sc(const struct qcs *qcs)
{
        return qcs->sd ? qcs->sd->sc : NULL;
}

/* Reset the <qcc> inactivity timeout for http-keep-alive timeout. */
static forceinline void qcc_reset_idle_start(struct qcc *qcc)
{
        qcc->idle_start = now_ms;
}

/* Return true if the mux timeout should be armed. */
static inline int qcc_may_expire(struct qcc *qcc)
{
        return !qcc->nb_sc;
}

/* Decrement <qcc> sc. */
static forceinline void qcc_rm_sc(struct qcc *qcc)
{
        BUG_ON(!qcc->nb_sc); /* Ensure sc count is always valid (ie >=0). */
        --qcc->nb_sc;
}

/* Checks if <qcc> connection can be used to attach new streams on it or if
 * reuse is definitely blocked. This is based on constant parameters such as a
 * connection error or timeout, the server max-reuse limit or if the peer has
 * requested a graceful shutdown. Flow control is not taken into account here
 * as it can be adjusted dynamically over the connection lifetime.
 *
 * Returns a boolean value indicating if reuse is possible.
 */
static int qcc_be_is_reusable(const struct qcc *qcc)
{
        const struct server *srv = __objt_server(qcc->conn->target);

        /* Connection on error or already on timeout. */
        if (qcc->flags & (QC_CF_ERR_CONN|QC_CF_ERRL) || !qcc->task)
                return 0;

        /* Shutdown initiated by the peer - in HTTP/3 this corresponds to a GOAWAY frame received. */
        if (qcc->flags & QC_CF_CONN_SHUT)
                return 0;

        if (srv->max_reuse >= 0 && qcc->tot_sc > srv->max_reuse)
                return 0;

        /* Ensure we do not exceed the maximum usable stream ID. */
        if (qcc->next_bidi_l > QCS_ID_MAX_STRM_CL_BIDI)
                return 0;

        return 1;
}

/* Indicates if a connection is idle and cannot be used anymore. If true the
 * connection should be released as soon as possible.
 */
static inline int qcc_is_dead(const struct qcc *qcc)
{
        /* Maintain connection if there is still request streams active. */
        if (qcc->nb_hreq)
                return 0;

        if (!conn_is_back(qcc->conn)) {
                /* FE conn considered dead if either :
                 * - transport or local error reported
                 * - MUX timeout expired
                 * - app layer shut
                 */
                return qcc->flags & (QC_CF_ERR_CONN|QC_CF_ERRL_DONE) ||
                       !qcc->task || qcc->app_st == QCC_APP_ST_SHUT;
        }
        else {
                /* BE conn considered dead if reuse is definitively blocked.
                 * Checks similar conditions as FE side and more.
                 */
                return !qcc_be_is_reusable(qcc);
        }
}

/* Refresh the timeout on <qcc> if needed depending on its state. */
static void qcc_refresh_timeout(struct qcc *qcc)
{
        const struct proxy *px = qcc->proxy;

        TRACE_ENTER(QMUX_EV_QCC_WAKE, qcc->conn);

        if (!qcc->task) {
                TRACE_DEVEL("already expired", QMUX_EV_QCC_WAKE, qcc->conn);
                goto leave;
        }

        /* Check if upper layer is responsible of timeout management. */
        if (!qcc_may_expire(qcc)) {
                TRACE_DEVEL("not eligible for timeout", QMUX_EV_QCC_WAKE, qcc->conn);
                qcc->task->expire = TICK_ETERNITY;
                task_queue(qcc->task);
                goto leave;
        }

        /* Timeout management
         * - detached streams with data left to send -> default timeout
         * - shutdown done -> timeout client-fin
         * - (FE only) stream waiting on incomplete request or no stream yet activated -> timeout http-request
         * - (FE only) idle after stream processing -> timeout http-keep-alive
         *
         * If proxy stop-stop in progress, immediate or spread close will be
         * processed if shutdown already one or connection is idle.
         */
        if (!LIST_ISEMPTY(&qcc->send_list) || !LIST_ISEMPTY(&qcc->tx.frms)) {
                TRACE_DEVEL("pending output data", QMUX_EV_QCC_WAKE, qcc->conn);
                qcc->task->expire = tick_add_ifset(now_ms, qcc->timeout);
        }
        else if (qcc->app_st >= QCC_APP_ST_SHUT) {
                TRACE_DEVEL("connection in closing", QMUX_EV_QCC_WAKE, qcc->conn);
                qcc->task->expire = tick_add_ifset(now_ms,
                                                   qcc->shut_timeout);
        }
        else if (!conn_is_back(qcc->conn)) {
                if (!LIST_ISEMPTY(&qcc->opening_list) || unlikely(!qcc->largest_bidi_r)) {
                        int timeout = px->timeout.httpreq;
                        struct qcs *qcs = NULL;
                        int base_time;

                        /* Use start time of first stream waiting on HTTP or
                         * qcc idle if no stream not yet used.
                         */
                        if (likely(!LIST_ISEMPTY(&qcc->opening_list)))
                                qcs = LIST_ELEM(qcc->opening_list.n, struct qcs *, el_opening);
                        base_time = qcs ? qcs->start : qcc->idle_start;

                        TRACE_DEVEL("waiting on http request", QMUX_EV_QCC_WAKE, qcc->conn, qcs);
                        qcc->task->expire = tick_add_ifset(base_time, timeout);
                }
                else {
                        /* Use http-request timeout if keep-alive timeout not set */
                        int timeout = tick_isset(px->timeout.httpka) ?
                                      px->timeout.httpka : px->timeout.httpreq;
                        TRACE_DEVEL("at least one request achieved but none currently in progress", QMUX_EV_QCC_WAKE, qcc->conn);
                        qcc->task->expire = tick_add_ifset(qcc->idle_start, timeout);
                }

                /* Fallback to client timeout if specific value not set. */
                if (!tick_isset(qcc->task->expire)) {
                        TRACE_DEVEL("fallback to default timeout", QMUX_EV_QCC_WAKE, qcc->conn);
                        qcc->task->expire = tick_add_ifset(now_ms, qcc->timeout);
                }
        }

        /* If proxy soft-stop in progress and connection is inactive,
         * close the connection immediately. If a close-spread-time is
         * configured, randomly spread the timer over a closing window.
         */
        if ((qcc->proxy->flags & (PR_FL_DISABLED|PR_FL_STOPPED)) &&
            !(global.tune.options & GTUNE_DISABLE_ACTIVE_CLOSE)) {

                /* Wake timeout task immediately if window already expired. */
                int remaining_window = tick_isset(global.close_spread_end) ?
                  tick_remain(now_ms, global.close_spread_end) : 0;

                TRACE_DEVEL("proxy disabled, prepare connection soft-stop", QMUX_EV_QCC_WAKE, qcc->conn);
                if (remaining_window) {
                        /* We don't need to reset the expire if it would
                         * already happen before the close window end.
                         */
                        if (!tick_isset(qcc->task->expire) ||
                            tick_is_le(global.close_spread_end, qcc->task->expire)) {
                                /* Set an expire value shorter than the current value
                                 * because the close spread window end comes earlier.
                                 */
                                qcc->task->expire = tick_add(now_ms,
                                                             statistical_prng_range(remaining_window));
                        }
                }
                else {
                        /* We are past the soft close window end, wake the timeout
                         * task up immediately.
                         */
                        qcc->task->expire = tick_add(now_ms, 0);
                        task_wakeup(qcc->task, TASK_WOKEN_TIMER);
                }
        }

        task_queue(qcc->task);

 leave:
        TRACE_LEAVE(QMUX_EV_QCS_NEW, qcc->conn);
}

/* Mark a stream as open if it was idle. This can be used on every
 * successful emission/reception operation to update the stream state.
 */
void qcs_idle_open(struct qcs *qcs)
{
        /* This operation must not be used if the stream is already closed. */
        BUG_ON_HOT(qcs->st == QC_SS_CLO);

        if (qcs->st == QC_SS_IDLE) {
                TRACE_STATE("opening stream", QMUX_EV_QCS_NEW, qcs->qcc->conn, qcs);
                qcs->st = QC_SS_OPEN;
        }
}

/* Close the local channel of <qcs> instance. */
void qcs_close_local(struct qcs *qcs)
{
        TRACE_STATE("closing stream locally", QMUX_EV_QCS_SEND, qcs->qcc->conn, qcs);

        /* The stream must have already been opened. */
        BUG_ON_HOT(qcs->st == QC_SS_IDLE);

        /* This operation cannot be used multiple times. */
        BUG_ON_HOT(qcs->st == QC_SS_HLOC || qcs->st == QC_SS_CLO);

        if (quic_stream_is_bidi(qcs->id)) {
                qcs->st = (qcs->st == QC_SS_HREM) ? QC_SS_CLO : QC_SS_HLOC;
        }
        else {
                /* Only local uni streams are valid for this operation. */
                BUG_ON_HOT(quic_stream_is_remote(qcs->qcc, qcs->id));
                qcs->st = QC_SS_CLO;
        }
}

/* Returns true if <qcs> can be purged. */
int qcs_is_completed(struct qcs *qcs)
{
        /* A stream is completed if fully closed and stconn released, or simply
         * detached and everything already sent.
         */
        return (qcs->st == QC_SS_CLO && !qcs_sc(qcs)) ||
               (qcs_is_close_local(qcs) && (qcs->flags & QC_SF_DETACH) &&
                !(qcs->flags & QC_SF_TO_STOP_SENDING));
}

/* Close the remote channel of <qcs> instance. */
static void qcs_close_remote(struct qcs *qcs)
{
        TRACE_STATE("closing stream remotely", QMUX_EV_QCS_RECV, qcs->qcc->conn, qcs);

        /* The stream must have already been opened. */
        BUG_ON_HOT(qcs->st == QC_SS_IDLE);

        /* This operation cannot be used multiple times. */
        BUG_ON_HOT(qcs->st == QC_SS_HREM || qcs->st == QC_SS_CLO);

        /* Ignore remote closure if QCS is already completed and thus already
         * scheduled for purging. This is possible as a detached QCS may be
         * considered as terminated even if only locally closed. This is
         * mandatory to avoid a double insert in purge list. This approach also
         * preserves QCS state on completion for better diagnostics.
         */
        if (qcs_is_completed(qcs)) {
                TRACE_STATE("ignore close remote on completed stream", QMUX_EV_QCS_RECV, qcs->qcc->conn, qcs);
                COUNT_IF(!(qcs->flags & QC_SF_DETACH), "QCS reported as completed prior remote closure but not detached.");
                COUNT_IF(!LIST_INLIST(&qcs->el_send),  "QCS reported as completed not in purge list");
                return;
        }

        if (quic_stream_is_bidi(qcs->id)) {
                qcs->st = (qcs->st == QC_SS_HLOC) ? QC_SS_CLO : QC_SS_HREM;
        }
        else {
                /* Only remote uni streams are valid for this operation. */
                BUG_ON_HOT(quic_stream_is_local(qcs->qcc, qcs->id));
                qcs->st = QC_SS_CLO;
        }

        /* Cancel STOP_SENDING emission as it is now unneeded. */
        if (qcs->st == QC_SS_CLO && (qcs->flags & QC_SF_TO_STOP_SENDING)) {
                qcs->flags &= ~QC_SF_TO_STOP_SENDING;
                /* Remove from send_list. Necessary to ensure BUG_ON() below is not triggered. */
                LIST_DEL_INIT(&qcs->el_send);
        }

        if (qcs_is_completed(qcs)) {
                BUG_ON(LIST_INLIST(&qcs->el_send));
                TRACE_STATE("add stream in purg_list", QMUX_EV_QCS_RECV, qcs->qcc->conn, qcs);
                LIST_APPEND(&qcs->qcc->purg_list, &qcs->el_send);
        }
}

int qcs_is_close_local(struct qcs *qcs)
{
        return qcs->st == QC_SS_HLOC || qcs->st == QC_SS_CLO;
}

int qcs_is_close_remote(struct qcs *qcs)
{
        return qcs->st == QC_SS_HREM || qcs->st == QC_SS_CLO;
}

/* Allocate if needed buffer <ncbuf> for stream <qcs>.
 *
 * Returns the buffer instance or NULL on allocation failure.
 */
static struct ncbuf *qcs_get_ncbuf(struct qcs *qcs, struct ncbuf *ncbuf)
{
        struct buffer buf = BUF_NULL;

        if (ncb_is_null(ncbuf)) {
                if (!b_alloc(&buf, DB_MUX_RX))
                        return NULL;

                *ncbuf = ncb_make(buf.area, buf.size, 0);
                ncb_init(ncbuf, 0);
        }

        return ncbuf;
}

/* Notify the stconn layer if initialized with TASK_WOKEN_MSG state and
 * eventually TASK_WOKEN_IO.
 */
static void qcs_alert(struct qcs *qcs)
{
        unsigned int state = TASK_WOKEN_MSG;

        TRACE_POINT(QMUX_EV_STRM_WAKE, qcs->qcc->conn, qcs);
        if (!qcs_sc(qcs))
                return;

        if (qcs->subs) {
                qcs->subs->events = 0;
                qcs->subs = NULL;
                state |= TASK_WOKEN_IO;
        }
        tasklet_wakeup(qcs_sc(qcs)->wait_event.tasklet, state);
}

int qcs_subscribe(struct qcs *qcs, int event_type, struct wait_event *es)
{
        struct qcc *qcc = qcs->qcc;

        TRACE_ENTER(QMUX_EV_STRM_SEND|QMUX_EV_STRM_RECV, qcc->conn, qcs);

        BUG_ON(event_type & ~(SUB_RETRY_SEND|SUB_RETRY_RECV));
        BUG_ON(qcs->subs && qcs->subs != es);

        es->events |= event_type;
        qcs->subs = es;

        if (event_type & SUB_RETRY_RECV)
                TRACE_DEVEL("subscribe(recv)", QMUX_EV_STRM_RECV, qcc->conn, qcs);

        if (event_type & SUB_RETRY_SEND)
                TRACE_DEVEL("subscribe(send)", QMUX_EV_STRM_SEND, qcc->conn, qcs);

        TRACE_LEAVE(QMUX_EV_STRM_SEND|QMUX_EV_STRM_RECV, qcc->conn, qcs);

        return 0;
}

void qcs_notify_recv(struct qcs *qcs)
{
        if (qcs->subs && qcs->subs->events & SUB_RETRY_RECV) {
                TRACE_POINT(QMUX_EV_STRM_WAKE, qcs->qcc->conn, qcs);
                tasklet_wakeup(qcs->subs->tasklet, TASK_WOKEN_IO);
                qcs->subs->events &= ~SUB_RETRY_RECV;
                if (!qcs->subs->events)
                        qcs->subs = NULL;
        }
}

void qcs_notify_send(struct qcs *qcs)
{
        if (qcs->subs && qcs->subs->events & SUB_RETRY_SEND) {
                TRACE_POINT(QMUX_EV_STRM_WAKE, qcs->qcc->conn, qcs);
                tasklet_wakeup(qcs->subs->tasklet, TASK_WOKEN_IO);
                qcs->subs->events &= ~SUB_RETRY_SEND;
                if (!qcs->subs->events)
                        qcs->subs = NULL;
        }
}

const struct buffer *qcs_tx_buf_const(const struct qcs *qcs)
{
        return conn_is_quic(qcs->qcc->conn) ?
          qc_stream_buf_get(qcs->tx.stream) : &qcs->tx.qmux_buf;
}

struct buffer *qcs_tx_buf(struct qcs *qcs)
{
        return conn_is_quic(qcs->qcc->conn) ?
          qc_stream_buf_get(qcs->tx.stream) : &qcs->tx.qmux_buf;
}

/* Returns total number of bytes not already sent to quic-conn layer. */
uint64_t qcs_prep_bytes(const struct qcs *qcs)
{
        const struct buffer *out = qcs_tx_buf_const(qcs);
        uint64_t diff, base_off;

        if (!out)
                return 0;

        if (conn_is_quic(qcs->qcc->conn)) {
                /* if ack_offset < buf_offset, it points to an older buffer. */
                base_off = MAX(qcs->tx.stream->buf_offset, qcs->tx.stream->ack_offset);
                diff = qcs->tx.fc.off_real - base_off;
                return b_data(out) - diff;
        }
        else {
                return b_data(out);
        }
}

/* Used as a callback for qc_stream_desc layer to notify about emission of a
 * STREAM frame of <data> length starting at <offset>.
 */
static void qcm_ctrl_send(struct qc_stream_desc *stream, uint64_t data, uint64_t offset)
{
        struct qcs *qcs = stream->ctx;
        struct qcc *qcc = qcs->qcc;
        uint64_t diff;

        TRACE_ENTER(QMUX_EV_QCS_SEND, qcc->conn, qcs);

        /* Real off MUST always be the greatest offset sent. */
        BUG_ON(offset > qcs->tx.fc.off_real);

        /* Check if the STREAM frame has already been notified. An empty FIN
         * frame must not be considered retransmitted.
         */
        if (data && offset + data <= qcs->tx.fc.off_real) {
                TRACE_DEVEL("offset already notified", QMUX_EV_QCS_SEND, qcc->conn, qcs);
                goto out;
        }

        /* An empty STREAM frame is only used to notify FIN. A retransmitted
         * empty FIN cannot be notified as QCS will be unsubscribed first.
         */
        BUG_ON(!data && !(qcs->flags & QC_SF_FIN_STREAM));

        qcs_idle_open(qcs);

        diff = offset + data - qcs->tx.fc.off_real;
        if (diff) {
                struct quic_fctl *fc_conn = &qcc->tx.fc;
                struct quic_fctl *fc_strm = &qcs->tx.fc;

                /* Ensure real offset never exceeds soft value. */
                BUG_ON(fc_conn->off_real + diff > fc_conn->off_soft);
                BUG_ON(fc_strm->off_real + diff > fc_strm->off_soft);

                /* increase offset sum on connection */
                if (qfctl_rinc(fc_conn, diff)) {
                        TRACE_STATE("connection flow-control reached",
                                    QMUX_EV_QCS_SEND, qcc->conn);
                }

                /* increase offset on stream */
                if (qfctl_rinc(fc_strm, diff)) {
                        TRACE_STATE("stream flow-control reached",
                                    QMUX_EV_QCS_SEND, qcc->conn, qcs);
                }
                /* Release buffer if everything sent and buf is full or stream is waiting for room. */
                if (!qcs_prep_bytes(qcs) &&
                    (b_full(&qcs->tx.stream->buf->buf) || qcs->flags & QC_SF_BLK_MROOM)) {
                        qc_stream_buf_release(qcs->tx.stream);
                        qcs->flags &= ~QC_SF_BLK_MROOM;
                        qcs_notify_send(qcs);
                }

                /* Add measurement for send rate. This is done at the MUX layer
                 * to account only for STREAM frames without retransmission.
                 */
                increment_send_rate(diff, 0);
        }

        if (!qc_stream_buf_get(qcs->tx.stream) || !qcs_prep_bytes(qcs)) {
                /* Remove stream from send_list if all was sent. */
                LIST_DEL_INIT(&qcs->el_send);
                TRACE_STATE("stream sent done", QMUX_EV_QCS_SEND, qcc->conn, qcs);

                /* Everything sent, close the stream if no new data will come.
                 * This is the case if either FIN has been sent or upper layer
                 * is detached.
                 *
                 * This is also performed if peer has resetted its input as no
                 * more transfer should occur on this QCS instance.
                 * TODO this assumption may be invalid depending on QUIC stream
                 * type - app proto should be responsible to deal with RS.
                 */
                if ((qcs->flags & (QC_SF_FIN_STREAM|QC_SF_DETACH)) ||
                    (!qcs_sc(qcs) && (qcs->flags & QC_SF_RECV_RESET))) {
                        /* Close stream locally. */
                        qcs_close_local(qcs);

                        if (qcs->flags & QC_SF_FIN_STREAM) {
                                qcs->tx.stream->flags |= QC_SD_FL_WAIT_FOR_FIN;
                                /* Reset flag to not emit multiple FIN STREAM frames. */
                                qcs->flags &= ~QC_SF_FIN_STREAM;
                        }

                        /* Unsubscribe from streamdesc when everything sent. */
                        qc_stream_desc_sub_send(qcs->tx.stream, NULL);

                        if (qcs_is_completed(qcs)) {
                                TRACE_STATE("add stream in purg_list", QMUX_EV_QCS_SEND, qcc->conn, qcs);
                                LIST_APPEND(&qcc->purg_list, &qcs->el_send);
                        }
                }
        }

 out:
        TRACE_LEAVE(QMUX_EV_QCS_SEND, qcc->conn, qcs);
}

/* Returns true if <qcc> buffer window does not have room for a new buffer. */
static inline int qcc_bufwnd_full(const struct qcc *qcc)
{
        const struct quic_conn *qc = qcc->conn->handle.qc;
        return qcc->tx.buf_in_flight >= qc->path->cwnd;
}

static void qcm_ctrl_room(struct qc_stream_desc *stream, uint64_t room)
{
        /* Context is different for active and released streams. */
        struct qcc *qcc = !(stream->flags & QC_SD_FL_RELEASE) ?
          ((struct qcs *)stream->ctx)->qcc : stream->ctx;
        qcc_notify_buf(qcc, room);
}

/* Report that one or several stream-desc buffers have been released for <qcc>
 * connection. <free_size> represent the sum of freed buffers sizes. May also
 * be used to notify about congestion window increase, in which case
 * <free_size> can be nul.
 */
void qcc_notify_buf(struct qcc *qcc, uint64_t free_size)
{
        struct qcs *qcs;

        TRACE_ENTER(QMUX_EV_QCC_WAKE, qcc->conn);

        /* Cannot have a negative buf_in_flight counter */
        BUG_ON(qcc->tx.buf_in_flight < free_size);
        qcc->tx.buf_in_flight -= free_size;

        if (qcc_bufwnd_full(qcc))
                return;

        if (qcc->flags & QC_CF_CONN_FULL) {
                TRACE_STATE("buf window now available", QMUX_EV_QCC_WAKE, qcc->conn);
                qcc->flags &= ~QC_CF_CONN_FULL;
        }

        /* TODO an optimization would be to only wake up a limited count of QCS
         * instances based on <free_size>. But it may not work if a woken QCS
         * is in error and does not try to allocate a buffer, leaving the
         * unwoken QCS indefinitely in the buflist.
         */
        while (!LIST_ISEMPTY(&qcc->buf_wait_list)) {
                qcs = LIST_ELEM(qcc->buf_wait_list.n, struct qcs *, el_buf);
                LIST_DEL_INIT(&qcs->el_buf);
                tot_time_stop(&qcs->timer.buf);
                qcs_notify_send(qcs);
        }

        TRACE_LEAVE(QMUX_EV_QCC_WAKE, qcc->conn);
}

/* A fatal error is detected locally for <qcc> connection. It should be closed
 * with a CONNECTION_CLOSE using <err> code. Set <app> to true to indicate that
 * the code must be considered as an application level error. This function
 * must not be called more than once by connection. If <tevt> is non null, it
 * is used as a connection level termination event code.
 */
void qcc_set_error(struct qcc *qcc, int err, int app, int tevt)
{
        /* This must not be called multiple times per connection. */
        BUG_ON(qcc->flags & QC_CF_ERRL);

        TRACE_STATE("connection on error", QMUX_EV_QCC_ERR, qcc->conn);

        qcc->flags |= QC_CF_ERRL;
        qcc->err = app ? quic_err_app(err) : quic_err_transport(err);

        if (tevt)
                qcc_report_term_evt(qcc, tevt);

        /* TODO
         * Ensure qcc_io_send() will be conducted to convert QC_CF_ERRL in
         * QC_CF_ERRL_DONE with CONNECTION_CLOSE frame emission. This may be
         * unnecessary if we are currently in the MUX tasklet context, but it
         * is too tedious too not forget a wakeup outside of this function for
         * the moment.
         */
        tasklet_wakeup(qcc->wait_event.tasklet);
}

void qcc_report_term_evt(struct qcc *qcc, enum muxc_term_event_type type)
{
        enum term_event_loc loc = tevt_loc_muxc;

        if (qcc->flags & QC_CF_IS_BACK)
                loc += 8;
        qcc->term_evts_log = tevt_report_event(qcc->term_evts_log, loc, type);
}

/* Increment glitch counter for <qcc> connection by <inc> steps. If configured
 * threshold reached, close the connection with an error code.
 */
int _qcc_report_glitch(struct qcc *qcc, int inc)
{
        const int max = QUIC_TUNE_FB_CONN_GET(sec_glitches_threshold, qcc->conn);

        qcc->glitches += inc;
        if (max && qcc->glitches >= max && !(qcc->flags & QC_CF_ERRL) &&
            (th_ctx->idle_pct <= global.tune.glitch_kill_maxidle)) {
                if (qcc->app_ops->report_susp) {
                        qcc->app_ops->report_susp(qcc->ctx);
                        qcc_set_error(qcc, qcc->err.code, 1,
                                      muxc_tevt_type_graceful_shut);
                }
                else {
                        qcc_set_error(qcc, QC_ERR_INTERNAL_ERROR, 0,
                                      muxc_tevt_type_internal_err);
                }
                return 1;
        }

        return 0;
}

/* Returns the number of streams which can still be opened until flow-control limit. */
int qcc_fctl_avail_streams(const struct qcc *qcc, int bidi)
{
        if (bidi) {
                const uint64_t next = qcc->next_bidi_l / 4;
                BUG_ON(qcc->rfctl.ms_bidi < next);
                return qcc->rfctl.ms_bidi - next;
        }
        else {
                const uint64_t next = qcc->next_uni_l / 4;
                BUG_ON(qcc->rfctl.ms_uni < next);
                return qcc->rfctl.ms_uni - next;
        }
}

/* Retrieves the maximum number of bidirectional remote streams that the peer
 * will be allowed to use during <conn> connection lifetime. This is guaranteed
 * to be a positive integer.
 */
static uint64_t qcc_max_strm_bidi_remote(const struct connection *conn)
{
        /* On FE side, streams may be limited by stream.max-total configuration. */
        if (!conn_is_back(conn) && quic_tune.fe.stream_max_total)
                return quic_tune.fe.stream_max_total;
        return (uint64_t)1 << 60;
}

/* Open a locally initiated stream for the connection <qcc>. Set <bidi> for a
 * bidirectional stream, else an unidirectional stream is opened. The next
 * available ID on the connection will be used according to the stream type.
 *
 * Returns the allocated stream instance or NULL on error.
 */
struct qcs *qcc_init_stream_local(struct qcc *qcc, int bidi)
{
        struct qcs *qcs;
        enum qcs_type type;
        uint64_t *next;

        TRACE_ENTER(QMUX_EV_QCS_NEW, qcc->conn);

        if (bidi) {
                /* Caller must ensure that max-streams peer flow-control won't be violated. */
                BUG_ON(qcc->rfctl.ms_bidi * 4 < qcc->next_bidi_l);
                next = &qcc->next_bidi_l;
                type = conn_is_back(qcc->conn) ? QCS_CLT_BIDI : QCS_SRV_BIDI;
        }
        else {
                BUG_ON(qcc->rfctl.ms_uni * 4 < qcc->next_uni_l);
                next = &qcc->next_uni_l;
                type = conn_is_back(qcc->conn) ? QCS_CLT_UNI : QCS_SRV_UNI;
        }

        qcs = qcs_new(qcc, *next, type);
        if (!qcs) {
                qcc_set_error(qcc, QC_ERR_INTERNAL_ERROR, 0,
                              muxc_tevt_type_internal_err);
                TRACE_DEVEL("leaving on error", QMUX_EV_QCS_NEW, qcc->conn);
                return NULL;
        }

        TRACE_PROTO("opening local stream",  QMUX_EV_QCS_NEW, qcc->conn, qcs);
        /* TODO emit STREAMS_BLOCKED if cannot create future streams. */
        *next += 4;

        TRACE_LEAVE(QMUX_EV_QCS_NEW, qcc->conn, qcs);
        return qcs;
}

/* Open a remote initiated stream for the connection <qcc> with ID <id>. The
 * caller is responsible to ensure that a stream with the same ID was not
 * already opened. This function will also create all intermediaries streams
 * with ID smaller than <id> not already opened before.
 *
 * Returns the allocated stream instance or NULL on error.
 */
static struct qcs *qcc_init_stream_remote(struct qcc *qcc, uint64_t id)
{
        struct qcs *qcs = NULL;
        enum qcs_type type;
        uint64_t *largest, max_id;

        TRACE_ENTER(QMUX_EV_QCS_NEW, qcc->conn);

        /* Function reserved to remote stream IDs. */
        BUG_ON(quic_stream_is_local(qcc, id));

        if (quic_stream_is_bidi(id)) {
                largest = &qcc->largest_bidi_r;
                type = conn_is_back(qcc->conn) ? QCS_SRV_BIDI : QCS_CLT_BIDI;
        }
        else {
                largest = &qcc->largest_uni_r;
                type = conn_is_back(qcc->conn) ? QCS_SRV_UNI : QCS_CLT_UNI;
        }

        /* RFC 9000 4.6. Controlling Concurrency
         *
         * An endpoint that receives a frame with a stream ID exceeding the
         * limit it has sent MUST treat this as a connection error of type
         * STREAM_LIMIT_ERROR
         */
        max_id = quic_stream_is_bidi(id) ? qcc->lfctl.ms_bidi * 4 :
                                           qcc->lfctl.ms_uni * 4;
        if (id >= max_id) {
                TRACE_ERROR("flow control error", QMUX_EV_QCS_NEW|QMUX_EV_PROTO_ERR, qcc->conn);
                qcc_set_error(qcc, QC_ERR_STREAM_LIMIT_ERROR, 0,
                              muxc_tevt_type_proto_err);
                goto err;
        }

        /* Only stream ID not already opened can be used. */
        BUG_ON(id < *largest);

        /* MAX_STREAMS emission must not allowed too big stream ID. */
        BUG_ON(*largest > QUIC_VARINT_8_BYTE_MAX);

        while (id >= *largest) {
                const char *str = *largest < id ? "initializing intermediary remote stream" :
                                                  "initializing remote stream";

                qcs = qcs_new(qcc, *largest, type);
                if (!qcs) {
                        TRACE_ERROR("stream fallocation failure", QMUX_EV_QCS_NEW, qcc->conn);
                        qcc_set_error(qcc, QC_ERR_INTERNAL_ERROR, 0,
                                      muxc_tevt_type_internal_err);
                        goto err;
                }

                TRACE_PROTO(str, QMUX_EV_QCS_NEW, qcc->conn, qcs);
                *largest += 4;
        }

 out:
        TRACE_LEAVE(QMUX_EV_QCS_NEW, qcc->conn, qcs);
        return qcs;

 err:
        TRACE_LEAVE(QMUX_EV_QCS_NEW, qcc->conn);
        return NULL;
}

/* Mark <qcs> as reserved for metadata transfer. As such, future txbuf
 * allocation won't be accounted against connection limit.
 */
void qcs_send_metadata(struct qcs *qcs)
{
        if (conn_is_quic(qcs->qcc->conn)) {
                /* Reserved for stream with Tx capability. */
                BUG_ON(!qcs->tx.stream);
                /* Cannot use if some data already transferred for this stream. */
                BUG_ON(qcs->tx.stream->ack_offset || !eb_is_empty(&qcs->tx.stream->buf_tree));

                qc_stream_desc_sub_room(qcs->tx.stream, NULL);
        }

        qcs->flags |= QC_SF_TXBUB_OOB;
}

/* Instantiate a stream and its associated stconn and sedesc. This is necessary
 * to transfer data after a new request reception. <buf> can be used to forward
 * the first received request data. <fin> must be set if the whole request is
 * already received.
 *
 * This function is only used on frontend side.
 *
 * Note that if <qcs> is already fully closed, nothing is instantiated. This is
 * useful if a RESET_STREAM was already emitted in response to a STOP_SENDING.
 *
 * Returns 0 on success else a negative error code. If <qcs> is already fully
 * closed and nothing is performed, it is considered as a success case.
 */
int qcs_attach_sc(struct qcs *qcs, struct buffer *buf, char fin)
{
        struct qcc *qcc = qcs->qcc;
        struct session *sess = qcc->conn->owner;

        TRACE_ENTER(QMUX_EV_STRM_RECV, qcc->conn, qcs);

        if (qcs->st == QC_SS_CLO) {
                TRACE_STATE("skip attach on already closed stream", QMUX_EV_STRM_RECV, qcc->conn, qcs);
                goto out;
        }

        /* TODO duplicated from mux_h2 */
        sess->t_idle = ns_to_ms(now_ns - sess->accept_ts) - sess->t_handshake;

        /* first create the stream endpoint descriptor */
        qcs->sd = sedesc_new();
        if (!qcs->sd) {
                TRACE_DEVEL("leaving on error", QMUX_EV_STRM_RECV, qcc->conn, qcs);
                return -1;
        }
        qcs->sd->se   = qcs;
        qcs->sd->conn = qcc->conn;
        se_fl_set(qcs->sd, SE_FL_T_MUX | SE_FL_ORPHAN | SE_FL_NOT_FIRST);
        se_expect_no_data(qcs->sd);

        if (!(global.tune.no_zero_copy_fwd & NO_ZERO_COPY_FWD_QUIC_SND))
                se_fl_set(qcs->sd, SE_FL_MAY_FASTFWD_CONS);

        if (!sc_new_from_endp(qcs->sd, sess, buf)) {
                TRACE_DEVEL("leaving on error", QMUX_EV_STRM_RECV, qcc->conn, qcs);
                return -1;
        }

        /* QCS must be identified as request stream prior to stconn instantiation. */
        BUG_ON(!(qcs->flags & QC_SF_HREQ_RECV));
        ++qcc->nb_sc;
        ++qcc->tot_sc;

        /* TODO duplicated from mux_h2 */
        sess->accept_date = date;
        sess->accept_ts   = now_ns;
        sess->t_handshake = 0;
        sess->t_idle = 0;

        /* A stream must have been registered for HTTP wait before attaching
         * it to sedesc. See <qcs_wait_http_req> for more info.
         */
        BUG_ON_HOT(!LIST_INLIST(&qcs->el_opening));
        LIST_DEL_INIT(&qcs->el_opening);

        /* rcv_buf may be skipped if request is wholly received on attach.
         * Ensure that similar flags are set for FIN both on rcv_buf and here.
         */
        if (fin) {
                TRACE_STATE("report end-of-input", QMUX_EV_STRM_RECV, qcc->conn, qcs);
                se_fl_set(qcs->sd, SE_FL_EOI);
                se_expect_data(qcs->sd);
        }

        /* A QCS can be already locally closed before stream layer
         * instantiation. This notably happens if STOP_SENDING was the first
         * frame received for this instance. In this case, an error is
         * immediately to the stream layer to prevent transmission.
         *
         * TODO it could be better to not instantiate at all the stream layer.
         * However, extra care is required to ensure QCS instance is released.
         */
        if (unlikely(qcs_is_close_local(qcs) || (qcs->flags & QC_SF_TO_RESET))) {
                TRACE_STATE("report early error", QMUX_EV_STRM_RECV, qcc->conn, qcs);
                se_fl_set_error(qcs->sd);
        }

        /* Graceful shutdown is initiated as soon as max stream is reached. */
        if (qcs->id == (qcc_max_strm_bidi_remote(qcc->conn) - 1) * 4) {
                TRACE_STATE("initiate shutdown as max remote bidi stream reached", QMUX_EV_STRM_RECV, qcc->conn, qcs);
                qcc_app_shutdown(qcc);
        }

 out:
        TRACE_LEAVE(QMUX_EV_STRM_RECV, qcc->conn, qcs);
        return 0;
}

/* Use this function for a stream <id> which is not in <qcc> stream tree. It
 * returns true if the associated stream is closed.
 */
static int qcc_stream_id_is_closed(struct qcc *qcc, uint64_t id)
{
        uint64_t *largest;

        /* This function must only be used for stream not present in the stream tree. */
        BUG_ON_HOT(eb64_lookup(&qcc->streams_by_id, id));

        if (quic_stream_is_local(qcc, id)) {
                largest = quic_stream_is_uni(id) ? &qcc->next_uni_l :
                                                   &qcc->next_bidi_l;
        }
        else {
                largest = quic_stream_is_uni(id) ? &qcc->largest_uni_r :
                                                   &qcc->largest_bidi_r;
        }

        return id < *largest;
}

/* Retrieve the stream instance from <id> ID. This can be used when receiving
 * STREAM, STREAM_DATA_BLOCKED, RESET_STREAM, MAX_STREAM_DATA or STOP_SENDING
 * frames. Set to false <receive_only> or <send_only> if these particular types
 * of streams are not allowed. If the stream instance is found, it is stored in
 * <out>.
 *
 * Returns 0 on success else non-zero. On error, a RESET_STREAM or a
 * CONNECTION_CLOSE is automatically emitted. Beware that <out> may be NULL
 * on success if the stream has already been closed.
 */
int qcc_get_qcs(struct qcc *qcc, uint64_t id, int receive_only, int send_only,
                struct qcs **out)
{
        struct eb64_node *node;

        TRACE_ENTER(QMUX_EV_QCC_RECV, qcc->conn);
        *out = NULL;

        if (!receive_only && quic_stream_is_uni(id) && quic_stream_is_remote(qcc, id)) {
                TRACE_ERROR("receive-only stream not allowed", QMUX_EV_QCC_RECV|QMUX_EV_QCC_NQCS|QMUX_EV_PROTO_ERR, qcc->conn, NULL, &id);
                qcc_set_error(qcc, QC_ERR_STREAM_STATE_ERROR, 0,
                              muxc_tevt_type_proto_err);
                goto err;
        }

        if (!send_only && quic_stream_is_uni(id) && quic_stream_is_local(qcc, id)) {
                TRACE_ERROR("send-only stream not allowed", QMUX_EV_QCC_RECV|QMUX_EV_QCC_NQCS|QMUX_EV_PROTO_ERR, qcc->conn, NULL, &id);
                qcc_set_error(qcc, QC_ERR_STREAM_STATE_ERROR, 0,
                              muxc_tevt_type_proto_err);
                goto err;
        }

        /* Search the stream in the connection tree. */
        node = eb64_lookup(&qcc->streams_by_id, id);
        if (node) {
                *out = eb64_entry(node, struct qcs, by_id);
                TRACE_DEVEL("using stream from connection tree", QMUX_EV_QCC_RECV, qcc->conn, *out);
                goto out;
        }

        /* Check if stream is already closed. */
        if (qcc_stream_id_is_closed(qcc, id)) {
                TRACE_DATA("already closed stream", QMUX_EV_QCC_RECV|QMUX_EV_QCC_NQCS, qcc->conn, NULL, &id);
                /* Consider this as a success even if <out> is left NULL. */
                goto out;
        }

        /* Create the stream. This is valid only for remote initiated one. A
         * local stream must have already been explicitly created by the
         * application protocol layer.
         */
        if (quic_stream_is_local(qcc, id)) {
                /* RFC 9000 19.8. STREAM Frames
                 *
                 * An endpoint MUST terminate the connection with error
                 * STREAM_STATE_ERROR if it receives a STREAM frame for a locally
                 * initiated stream that has not yet been created, or for a send-only
                 * stream.
                 */
                TRACE_ERROR("locally initiated stream not yet created", QMUX_EV_QCC_RECV|QMUX_EV_QCC_NQCS|QMUX_EV_PROTO_ERR, qcc->conn, NULL, &id);
                qcc_set_error(qcc, QC_ERR_STREAM_STATE_ERROR, 0,
                              muxc_tevt_type_proto_err);
                goto err;
        }
        else {
                /* Remote stream not found - try to open it. */
                *out = qcc_init_stream_remote(qcc, id);
                if (!*out) {
                        TRACE_ERROR("stream creation error", QMUX_EV_QCC_RECV|QMUX_EV_QCC_NQCS, qcc->conn, NULL, &id);
                        goto err;
                }
        }

 out:
        TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn, *out);
        return 0;

 err:
        TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
        return 1;
}

/* Convert <b> out-of-order storage into a contiguous buffer. */
static inline struct buffer qcs_b_dup(const struct qc_stream_rxbuf *b)
{
        if (b) {
                const struct ncbuf *ncb = &b->ncb;
                return b_make(ncb_orig(ncb), ncb->size, ncb->head, ncb_data(ncb, 0));
        }
        else {
                return BUF_NULL;
        }
}

/* Transfer data into <qcs> stream <rxbuf> current Rx buffer from its directly
 * following buffer. This is useful when parsing was interrupted due to partial
 * data. If following buffer does not exists, nothing is done.
 *
 * Returns 0 if data transfer was performed.
 */
static int qcs_transfer_rx_data(struct qcs *qcs, struct qc_stream_rxbuf *rxbuf)
{
        struct qc_stream_rxbuf *rxbuf_next;
        struct eb64_node *next;
        struct buffer b, b_next;
        enum ncb_ret ncb_ret;
        size_t to_copy;
        int ret = 1;

        BUG_ON(ncb_is_full(&rxbuf->ncb));

        next = eb64_next(&rxbuf->off_node);
        if (!next)
                goto out;

        rxbuf_next = container_of(next, struct qc_stream_rxbuf, off_node);
        if (rxbuf_next->off_node.key == rxbuf->off_end &&
            ncb_data(&rxbuf_next->ncb, 0)) {
                eb64_delete(&rxbuf->off_node);
                eb64_delete(next);

                b = qcs_b_dup(rxbuf);
                b_next = qcs_b_dup(rxbuf_next);
                to_copy = MIN(b_data(&b_next), ncb_size(&rxbuf->ncb) - b_data(&b));

                ncb_ret = ncb_add(&rxbuf->ncb, ncb_data(&rxbuf->ncb, 0),
                                  b_head(&b_next), to_copy, NCB_ADD_COMPARE);
                BUG_ON(ncb_ret != NCB_RET_OK);

                ncb_ret = ncb_advance(&rxbuf_next->ncb, to_copy);
                BUG_ON(ncb_ret != NCB_RET_OK);

                rxbuf->off_node.key = qcs->rx.offset;
                rxbuf->off_end = qcs->rx.offset + b_data(&b) + to_copy;
                eb64_insert(&qcs->rx.bufs, &rxbuf->off_node);

                /* Increment next rxbuf offset. This must not exceed off_end. */
                rxbuf_next->off_node.key += to_copy;
                BUG_ON(rxbuf_next->off_node.key > rxbuf_next->off_end);
                /* Now reinsert next rxbuf unless it has been completely truncated. */
                if (rxbuf_next->off_node.key < rxbuf_next->off_end) {
                        eb64_insert(&qcs->rx.bufs, &rxbuf_next->off_node);
                }
                else {
                        b_free(&b_next);
                        offer_buffers(NULL, 1);
                        pool_free(pool_head_qc_stream_rxbuf, rxbuf_next);
                        bdata_ctr_bdec(&qcs->rx.data);
                }

                ret = 0;
        }

 out:
        return ret;
}

/* Returns the Rx buffer instance for <qcs> stream read offset. May be NULL if
 * not already allocated.
 */
static struct qc_stream_rxbuf *qcs_get_curr_rxbuf(struct qcs *qcs)
{
        struct eb64_node *node;
        struct qc_stream_rxbuf *buf;

        node = eb64_first(&qcs->rx.bufs);
        if (!node)
                return NULL;

        buf = container_of(node, struct qc_stream_rxbuf, off_node);
        if (qcs->rx.offset < buf->off_node.key) {
                /* first buffer allocated for a future offset */
                return NULL;
        }

        /* Ensures obsolete buffer are not kept inside QCS */
        BUG_ON(buf->off_end < qcs->rx.offset);
        return buf;
}

/* Returns the amount of data readable at <qcs> stream on current buffer. Note
 * that this does account for hypothetical contiguous data divided on other
 * Rx buffers instances.
 */
static ncb_sz_t qcs_rx_avail_data(struct qcs *qcs)
{
        struct qc_stream_rxbuf *b = qcs_get_curr_rxbuf(qcs);
        return b ? ncb_data(&b->ncb, 0) : 0;
}

/* Account for <bytes> in <qcc> connection flow-control. If needed, a MAX_DATA
 * frame is prepared for emission.
 */
static void qcc_account_for_conn_fctl(struct qcc *qcc, uint64_t bytes)
{
        struct quic_frame *frm;

        TRACE_ENTER(QMUX_EV_QCC_RECV, qcc->conn);

        qcc->lfctl.offsets_consume += bytes;
        if (qcc->lfctl.md - qcc->lfctl.offsets_consume < qcc->lfctl.md_init / 2) {
                TRACE_DATA("increase conn credit via MAX_DATA", QMUX_EV_QCS_RECV, qcc->conn);
                frm = qc_frm_alloc(QUIC_FT_MAX_DATA);
                if (!frm) {
                        qcc_set_error(qcc, QC_ERR_INTERNAL_ERROR, 0,
                                      muxc_tevt_type_internal_err);
                        goto out;
                }

                qcc->lfctl.md = qcc->lfctl.offsets_consume + qcc->lfctl.md_init;
                frm->max_data.max_data = qcc->lfctl.md;

                LIST_APPEND(&qcc->lfctl.frms, &frm->list);
                tasklet_wakeup(qcc->wait_event.tasklet);
        }

 out:
        TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
}

/* Remove <bytes> from <buf> current Rx buffer of <qcs> stream. Flow-control
 * for received offsets may be allocated for the peer if needed.
 */
static void qcs_consume(struct qcs *qcs, uint64_t bytes, struct qc_stream_rxbuf *buf)
{
        struct qcc *qcc = qcs->qcc;
        struct quic_frame *frm;
        enum ncb_ret ret;
        uint64_t diff, inc = 0;

        TRACE_ENTER(QMUX_EV_QCS_RECV, qcc->conn, qcs);

        /* <buf> must be current QCS Rx buffer. */
        BUG_ON_HOT(buf->off_node.key > qcs->rx.offset ||
                   qcs->rx.offset >= buf->off_end);

        ret = ncb_advance(&buf->ncb, bytes);
        if (ret) {
                ABORT_NOW(); /* should not happens because removal only in data */
        }

        qcs->rx.offset += bytes;
        /* QCS Rx offset must only point directly up to the next buffer. */
        BUG_ON_HOT(qcs->rx.offset > buf->off_end);

        /* Not necessary to emit a MAX_STREAM_DATA if all data received. */
        if (qcs->flags & QC_SF_SIZE_KNOWN)
                goto conn_fctl;

        /* Check if a MAX_STREAM_DATA frame should be emitted, determined by
         * the consumed capacity. If no more than 2 Rx buffers can be allocated
         * per QCS, the limit is set to half the capacity. Else, the limit is
         * set to match bufsize.
         */
        if (qcs->rx.msd - qcs->rx.msd_base < qcm_stream_rx_bufsz() * 2) {
                if ((qcs->rx.offset - qcs->rx.msd_base) * 2 >= qcs->rx.msd - qcs->rx.msd_base)
                        inc = qcs->rx.offset - qcs->rx.msd_base;
        }
        else {
                diff = qcs->rx.offset - qcs->rx.msd_base;
                while (diff >= qcm_stream_rx_bufsz()) {
                        inc += qcm_stream_rx_bufsz();
                        diff -= qcm_stream_rx_bufsz();
                }
        }

        if (inc) {
                frm = qcs->tx.msd_frm;
                if (!frm) {
                        frm = qc_frm_alloc(QUIC_FT_MAX_STREAM_DATA);
                        if (!frm) {
                                qcc_set_error(qcc, QC_ERR_INTERNAL_ERROR, 0,
                                              muxc_tevt_type_internal_err);
                                return;
                        }

                        frm->max_stream_data.id = qcs->id;
                        LIST_APPEND(&qcc->lfctl.frms, &frm->list);
                        qcs->tx.msd_frm = frm;
                }

                TRACE_DATA("increase stream credit via MAX_STREAM_DATA", QMUX_EV_QCS_RECV, qcc->conn, qcs);
                qcs->rx.msd += inc;
                qcs->rx.msd_base += inc;

                frm->max_stream_data.max_stream_data = qcs->rx.msd;

                tasklet_wakeup(qcc->wait_event.tasklet);
        }

 conn_fctl:
        qcc_account_for_conn_fctl(qcc, bytes);

        TRACE_LEAVE(QMUX_EV_QCS_RECV, qcc->conn, qcs);
}

/* Decode the content of STREAM frames already received on the stream instance
 * <qcs> from the <qcc> connection.
 *
 * It is safe to remove <qcs> from <qcc> recv_list after decoding is done. Even
 * if an error is returned, caller should consider that no further Rx
 * processing can be performed for the stream, until new bytes are available.
 *
 * Returns the result of app_ops rcv_buf callback, which is the number of bytes
 * successfully transcoded, or a negative error code. If no error occurred but
 * decoding cannot proceed due to missing data, the return value is 0. The
 * value 0 may also be returned when dealing with a standalone FIN signal.
 */
static int qcc_decode_qcs(struct qcc *qcc, struct qcs *qcs)
{
        struct qc_stream_rxbuf *rxbuf;
        struct buffer b;
        ssize_t ret;
        int fin = 0;
        int prev_glitches = qcc->glitches;

        TRACE_ENTER(QMUX_EV_QCS_RECV, qcc->conn, qcs);

        if (qcc->flags & QC_CF_ERRL) {
                TRACE_DATA("connection on error", QMUX_EV_QCC_RECV, qcc->conn);
                ret = -1;
                goto err;
        }

 restart:
        rxbuf = qcs_get_curr_rxbuf(qcs);
        b = qcs_b_dup(rxbuf);

        /* Signal FIN to application if STREAM FIN received with all data. */
        if (qcs_is_close_remote(qcs))
                fin = 1;

        if (!(qcs->flags & QC_SF_READ_ABORTED)) {
                ret = qcc->app_ops->rcv_buf(qcs, &b, fin);

                if (qcc->glitches != prev_glitches && !(qcc->flags & QC_CF_IS_BACK))
                        session_add_glitch_ctr(qcc->conn->owner, qcc->glitches - prev_glitches);

                if (ret < 0) {
                        TRACE_ERROR("decoding error", QMUX_EV_QCS_RECV, qcc->conn, qcs);
                        goto err;
                }

                /* App layer cannot decode due to partial data, which is stored
                 * at a Rx buffer boundary. Try to realign data if possible and
                 * restart decoding.
                 */
                if (!ret && rxbuf && !(qcs->flags & QC_SF_DEM_FULL) &&
                    qcs->rx.offset + ncb_data(&rxbuf->ncb, 0) == rxbuf->off_end) {
                        if (!qcs_transfer_rx_data(qcs, rxbuf)) {
                                TRACE_DEVEL("restart parsing after data realignment", QMUX_EV_QCS_RECV, qcc->conn, qcs);
                                goto restart;
                        }
                }

                if (qcs->flags & QC_SF_TO_RESET) {
                        if (qcs_sc(qcs) && !se_fl_test(qcs->sd, SE_FL_ERROR|SE_FL_ERR_PENDING)) {
                                se_fl_set_error(qcs->sd);
                                qcs_alert(qcs);
                        }
                }
        }
        else {
                TRACE_DATA("ignore read on stream", QMUX_EV_QCS_RECV, qcc->conn, qcs);
                ret = b_data(&b);
        }

        if (rxbuf) {
                if (ret)
                        qcs_consume(qcs, ret, rxbuf);

                if (ncb_is_empty(&rxbuf->ncb)) {
                        qcs_free_rxbuf(qcs, rxbuf);

                        /* Close QCS remotely if only one Rx buffer remains and
                         * all data with FIN already stored in it. This is
                         * necessary to be performed before app_ops rcv_buf to
                         * ensure FIN is correctly signalled.
                         */
                        if (qcs->flags & QC_SF_SIZE_KNOWN && !eb_is_empty(&qcs->rx.bufs)) {
                                const ncb_sz_t avail = qcs_rx_avail_data(qcs);
                                if (qcs->rx.offset + avail == qcs->rx.offset_max)
                                        qcs_close_remote(qcs);
                        }
                }
        }

        if (ret || (!b_data(&b) && fin))
                qcs_notify_recv(qcs);

        TRACE_LEAVE(QMUX_EV_QCS_RECV, qcc->conn, qcs);
        return ret;

 err:
        TRACE_DEVEL("leaving on error", QMUX_EV_QCS_RECV, qcc->conn, qcs);
        return ret;
}

/* Allocate if needed and retrieve <qcs> stream buffer for data reception.
 *
 * Returns buffer pointer. May be NULL on allocation failure.
 */
struct buffer *qcc_get_stream_rxbuf(struct qcs *qcs)
{
        return b_alloc(&qcs->rx.app_buf, DB_MUX_RX);
}

/* Allocate if needed and retrieve <qcs> stream buffer for data emission.
 *
 * <err> is an output argument which is useful to differentiate the failure
 * cause when the buffer cannot be allocated. It is set to 0 if the connection
 * buffer window is full. For fatal errors, its value is non-zero.
 *
 * Streams reserved for application protocol metadata transfer are not subject
 * to the buffer limit per connection. Hence, for them only a memory error
 * can prevent a buffer allocation.
 *
 * Returns buffer pointer. May be NULL on allocation failure, in which case
 * <err> will refer to the cause.
 */
struct buffer *qcc_get_stream_txbuf(struct qcs *qcs, int *err, int small)
{
        struct qcc *qcc = qcs->qcc;
        struct buffer *out = qcs_tx_buf(qcs);

        /* Stream must not try to reallocate a buffer if currently waiting for one. */
        BUG_ON(LIST_INLIST(&qcs->el_buf));

        *err = 0;

        if (conn_is_quic(qcc->conn) && !out) {
                if (likely(!(qcs->flags & QC_SF_TXBUB_OOB))) {
                        if ((qcc->flags & QC_CF_CONN_FULL)) {
                                LIST_APPEND(&qcc->buf_wait_list, &qcs->el_buf);
                                tot_time_start(&qcs->timer.buf);
                                goto out;
                        }

                        if (qcc_bufwnd_full(qcc)) {
                                TRACE_STATE("no more room", QMUX_EV_QCS_SEND, qcc->conn, qcs);
                                LIST_APPEND(&qcc->buf_wait_list, &qcs->el_buf);
                                tot_time_start(&qcs->timer.buf);
                                qcc->flags |= QC_CF_CONN_FULL;
                                goto out;
                        }
                }

                out = qc_stream_buf_alloc(qcs->tx.stream, qcs->tx.fc.off_real, small);
                if (!out) {
                        TRACE_ERROR("stream desc alloc failure", QMUX_EV_QCS_SEND, qcc->conn, qcs);
                        *err = 1;
                        goto out;
                }

                if (likely(!(qcs->flags & QC_SF_TXBUB_OOB)))
                        qcc->tx.buf_in_flight += b_size(out);
        }
        else if (!conn_is_quic(qcc->conn)) {
                if ((small && !b_alloc_small(out)) ||
                    (!small && !b_alloc(out, DB_MUX_TX))) {
                        TRACE_ERROR("QCS tx buf alloc failure", QMUX_EV_QCS_SEND, qcc->conn, qcs);
                        *err = 1;
                        goto out;
                }
        }

 out:
        return out;
}

/* Reallocate <qcs> stream buffer to convert a small buffer to a bigger one.
 * Contrary to standard allocation, this function will never stop due to a full
 * buffer window. The smaller buffer is released first which guarantees that the
 * buffer window has room left.
 *
 * Returns buffer pointer or NULL on allocation failure.
 */
struct buffer *qcc_realloc_stream_txbuf(struct qcs *qcs)
{
        struct qcc *qcc = qcs->qcc;
        struct buffer *out = qcs_tx_buf(qcs);

        /* Stream must not try to reallocate a buffer if currently waiting for one. */
        BUG_ON(LIST_INLIST(&qcs->el_buf));

        if (conn_is_quic(qcc->conn)) {
                if (likely(!(qcs->flags & QC_SF_TXBUB_OOB))) {
                        /* Reduce buffer window. As such there is always some space
                         * left for a new buffer allocation.
                         */
                        BUG_ON(qcc->tx.buf_in_flight < b_size(out));
                        qcc->tx.buf_in_flight -= b_size(out);
                }

                out = qc_stream_buf_realloc(qcs->tx.stream);
                if (!out) {
                        TRACE_ERROR("buffer alloc failure", QMUX_EV_QCS_SEND, qcc->conn, qcs);
                        goto out;
                }

                if (likely(!(qcs->flags & QC_SF_TXBUB_OOB)))
                        qcc->tx.buf_in_flight += b_size(out);
        }
        else {
                b_free(out);
                if (!b_alloc(out, DB_MUX_TX)) {
                        TRACE_ERROR("QCS tx buf alloc failure", QMUX_EV_QCS_SEND, qcc->conn, qcs);
                        goto out;
                }
        }

 out:
        return out && b_size(out) ? out : NULL;
}

/* Try to realign <out> buffer for <qcs> stream. This is done only if there is
 * no data waiting for ACK.
 *
 * Returns 0 if realign was performed else non-zero.
 */
int qcc_realign_stream_txbuf(const struct qcs *qcs, struct buffer *out)
{
        if (qcs_prep_bytes(qcs) == b_data(out)) {
                b_slow_realign(out, trash.area, b_data(out));
                return 0;
        }

        return 1;
}

/* Release the current <qcs> Tx buffer. This is useful if space left is not
 * enough anymore. A new instance can then be allocated to continue sending.
 *
 * This operation fails if there is not yet sent bytes in the buffer. In this
 * case, stream layer should interrupt sending until further notification.
 *
 * Returns 0 if buffer is released and a new one can be allocated or non-zero
 * if there is still remaining data.
 */
int qcc_release_stream_txbuf(struct qcs *qcs)
{
        const uint64_t bytes = qcs_prep_bytes(qcs);

        /* Cannot release buffer if prepared data is not fully sent. */
        if (bytes) {
                qcs->flags |= QC_SF_BLK_MROOM;
                return 1;
        }

        qc_stream_buf_release(qcs->tx.stream);
        return 0;
}

/* Returns true if stream layer can proceed to emission via <qcs>. */
int qcc_stream_can_send(const struct qcs *qcs)
{
        return !(qcs->flags & QC_SF_BLK_MROOM) && !LIST_INLIST(&qcs->el_buf);
}

/* Wakes up every streams of <qcc> which are currently waiting for sending but
 * are blocked on connection flow control.
 */
static void qcc_notify_fctl(struct qcc *qcc)
{
        struct qcs *qcs;

        while (!LIST_ISEMPTY(&qcc->fctl_list)) {
                qcs = LIST_ELEM(qcc->fctl_list.n, struct qcs *, el_fctl);
                LIST_DEL_INIT(&qcs->el_fctl);
                tot_time_stop(&qcs->timer.fctl);
                qcs_notify_send(qcs);
        }
}

/* Free <qcc> STREAM frames in Tx list. */
static void qcc_clear_frms(struct qcc *qcc)
{
        TRACE_STATE("resetting STREAM frames list", QMUX_EV_QCC_SEND, qcc->conn);
        while (!LIST_ISEMPTY(&qcc->tx.frms)) {
                struct quic_frame *frm = LIST_ELEM(qcc->tx.frms.n, struct quic_frame *, list);
                qc_frm_free(conn_is_quic(qcc->conn) ? qcc->conn->handle.qc : NULL, &frm);
        }
}

/* Register <qcs> stream for emission of STREAM, STOP_SENDING or RESET_STREAM.
 * Set <urg> to true if stream should be emitted in priority. This is useful
 * when sending STOP_SENDING or RESET_STREAM, or for emission on an application
 * control stream.
 */
static void _qcc_send_stream(struct qcs *qcs, int urg)
{
        struct qcc *qcc = qcs->qcc;

        qcc_clear_frms(qcc);

        if (urg) {
                /* qcc_emit_rs_ss() relies on reset/aborted streams in send_list front. */
                BUG_ON(!(qcs->flags & (QC_SF_TO_RESET|QC_SF_TO_STOP_SENDING|QC_SF_TXBUB_OOB)));

                LIST_DEL_INIT(&qcs->el_send);
                LIST_INSERT(&qcc->send_list, &qcs->el_send);
        }
        else {
                if (!LIST_INLIST(&qcs->el_send))
                        LIST_APPEND(&qcs->qcc->send_list, &qcs->el_send);
        }
}

/* Prepare for the emission of RESET_STREAM on <qcs> with error code <err>. If
 * <tevt> is non null, it is used as a stream level termination event code.
 */
void qcc_reset_stream(struct qcs *qcs, int err, int tevt)
{
        struct qcc *qcc = qcs->qcc;
        const uint64_t diff = qcs_prep_bytes(qcs);

        if ((qcs->flags & QC_SF_TO_RESET) || qcs_is_close_local(qcs))
                return;

        /* TODO if QCS waiting for buffer, it could be removed from
         * <qcc.buf_wait_list> if sending is closed now.
         */

        TRACE_STATE("reset stream", QMUX_EV_QCS_END, qcc->conn, qcs);
        qcs->flags |= QC_SF_TO_RESET;
        qcs->rs_err = err;

        /* On BE side, a QCS may be resetted before any data emission. */
        if (conn_is_back(qcs->qcc->conn))
                qcs_idle_open(qcs);

        if (diff) {
                const int soft_blocked = qfctl_sblocked(&qcc->tx.fc);

                /* Soft offset cannot be inferior to real one. */
                BUG_ON(qcc->tx.fc.off_soft - diff < qcc->tx.fc.off_real);

                /* Subtract to conn flow control data amount prepared on stream not yet sent. */
                qcc->tx.fc.off_soft -= diff;
                if (soft_blocked && !qfctl_sblocked(&qcc->tx.fc))
                        qcc_notify_fctl(qcc);

                /* Reset QCS soft off to prevent BUG_ON() on qcs_destroy(). */
                qcs->tx.fc.off_soft = qcs->tx.fc.off_real;
        }

        /* Report send error to stream-endpoint layer. */
        if (qcs_sc(qcs)) {
                se_fl_set_error(qcs->sd);
                if (tevt)
                        se_report_term_evt(qcs->sd, tevt);
                qcs_alert(qcs);
        }

        _qcc_send_stream(qcs, 1);
        tasklet_wakeup(qcc->wait_event.tasklet);
}

/* Register <qcs> stream for STREAM emission. Set <urg> to 1 if stream content
 * should be treated in priority compared to other streams. <count> must
 * contains the size of the frame payload, used for flow control accounting.
 */
void qcc_send_stream(struct qcs *qcs, int urg, int count)
{
        struct qcc *qcc = qcs->qcc;

        TRACE_ENTER(QMUX_EV_QCS_SEND, qcc->conn, qcs);

        /* Cannot send STREAM if already closed. */
        BUG_ON(qcs_is_close_local(qcs));

        _qcc_send_stream(qcs, urg);

        if (count) {
                qfctl_sinc(&qcc->tx.fc, count);
                qfctl_sinc(&qcs->tx.fc, count);
                if (conn_is_quic(qcc->conn))
                        bdata_ctr_add(&qcs->tx.stream->data, count);
        }

        TRACE_LEAVE(QMUX_EV_QCS_SEND, qcc->conn, qcs);
}

/* Prepare for the emission of STOP_SENDING on <qcs>. */
void qcc_abort_stream_read(struct qcs *qcs, int err)
{
        struct qcc *qcc = qcs->qcc;

        TRACE_ENTER(QMUX_EV_QCC_NEW, qcc->conn, qcs);

        if ((qcs->flags & QC_SF_READ_ABORTED) || qcs_is_close_remote(qcs))
                goto end;

        TRACE_STATE("abort stream read", QMUX_EV_QCS_END, qcc->conn, qcs);
        qcs->flags |= (QC_SF_TO_STOP_SENDING|QC_SF_READ_ABORTED);
        qcs->ss_err = err;

        _qcc_send_stream(qcs, 1);
        tasklet_wakeup(qcc->wait_event.tasklet);

 end:
        TRACE_LEAVE(QMUX_EV_QCC_NEW, qcc->conn, qcs);
}

/* Install the applicative layer of a QUIC connection on mux <qcc>.
 * Returns 0 on success else non-zero.
 */
int qcc_install_app_ops(struct qcc *qcc)
{
        struct connection *conn = qcc->conn;
        const struct qcc_app_ops *app_ops;
        struct server *srv;
        const char *alpn;
        int alpn_len;

        TRACE_ENTER(QMUX_EV_QCC_NEW, conn);

        if (!conn_get_alpn(conn, &alpn, &alpn_len)) {
                if (!conn_is_back(conn))
                        goto err;

                srv = __objt_server(conn->target);
                HA_RWLOCK_RDLOCK(SERVER_LOCK, &srv->path_params.param_lock);
                alpn = srv->path_params.nego_alpn;
                HA_RWLOCK_RDUNLOCK(SERVER_LOCK, &srv->path_params.param_lock);

                if (!alpn)
                        goto err;
                alpn_len = strlen(srv->path_params.nego_alpn);
        }

        app_ops = quic_alpn_to_app_ops(alpn, alpn_len);
        if (!app_ops)
                goto err;

        if (app_ops->init && !app_ops->init(qcc)) {
                TRACE_ERROR("application layer install error", QMUX_EV_QCC_NEW, conn);
                goto err;
        }

        TRACE_PROTO("application layer installed", QMUX_EV_QCC_NEW, conn);
        qcc->app_ops = app_ops;

        TRACE_LEAVE(QMUX_EV_QCC_NEW, conn);
        return 0;

 err:
        TRACE_LEAVE(QMUX_EV_QCC_NEW, conn);
        return 1;
}

/* Retrieves the Rx buffer instance usable to store STREAM data starting at
 * <offset>. It is dynamically allocated if not already instantiated. <len>
 * must contains the size of the STREAM frame. It may be reduced by the
 * function if data is too large relative to the buffer starting offset.
 * Another buffer instance should be allocated to store the remaining data.
 *
 * Returns the buffer instance or NULL in case of error.
 */
static struct qc_stream_rxbuf *qcs_get_rxbuf(struct qcs *qcs, uint64_t offset,
                                             uint64_t *len)
{
        struct qcc *qcc = qcs->qcc;
        struct eb64_node *node;
        struct qc_stream_rxbuf *buf;
        struct ncbuf *ncbuf;

        TRACE_ENTER(QMUX_EV_QCS_RECV, qcs->qcc->conn, qcs);

        node = eb64_lookup_le(&qcs->rx.bufs, offset);
        if (node)
                buf = container_of(node, struct qc_stream_rxbuf, off_node);

        if (!node || offset >= buf->off_end) {
                const uint64_t aligned_off = offset - (offset % qcm_stream_rx_bufsz());

                TRACE_DEVEL("allocating a new entry", QMUX_EV_QCS_RECV, qcs->qcc->conn, qcs);
                buf = pool_alloc(pool_head_qc_stream_rxbuf);
                if (!buf) {
                        TRACE_ERROR("qcs rxbuf alloc error", QMUX_EV_QCC_RECV, qcc->conn, qcs);
                        goto err;
                }

                buf->ncb = NCBUF_NULL;
                buf->off_node.key = aligned_off;
                buf->off_end = aligned_off + qcm_stream_rx_bufsz();
                eb64_insert(&qcs->rx.bufs, &buf->off_node);
                bdata_ctr_binc(&qcs->rx.data);
        }

        ncbuf = &buf->ncb;
        if (!qcs_get_ncbuf(qcs, ncbuf) || ncb_is_null(ncbuf)) {
                TRACE_ERROR("receive ncbuf alloc failure", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV, qcc->conn, qcs);
                goto err;
        }

        if (offset + *len > buf->off_end)
                *len = buf->off_end - offset;

        TRACE_LEAVE(QMUX_EV_QCS_RECV, qcs->qcc->conn, qcs);
        return buf;

 err:
        TRACE_DEVEL("leaving on error", QMUX_EV_QCS_RECV, qcs->qcc->conn, qcs);
        return NULL;
}

/* Handle a new STREAM frame for stream with id <id>. Payload is pointed by
 * <data> with length <len> and represents the offset <offset>. <fin> is set if
 * the QUIC frame FIN bit is set.
 *
 * Returns 0 on success else non-zero. On error, the received frame should not
 * be acknowledged.
 */
int qcc_recv(struct qcc *qcc, uint64_t id, uint64_t len, uint64_t offset,
             char fin, char *data)
{
        const int fin_standalone = (!len && fin);
        struct qcs *qcs;
        enum ncb_ret ncb_ret;
        uint64_t left;
        int ret;

        TRACE_ENTER(QMUX_EV_QCC_RECV, qcc->conn);

        if (qcc->flags & QC_CF_ERRL) {
                TRACE_DATA("connection on error", QMUX_EV_QCC_RECV, qcc->conn);
                goto err;
        }

        /* RFC 9000 19.8. STREAM Frames
         *
         * An endpoint MUST terminate the connection with error
         * STREAM_STATE_ERROR if it receives a STREAM frame for a locally
         * initiated stream that has not yet been created, or for a send-only
         * stream.
         */
        if (qcc_get_qcs(qcc, id, 1, 0, &qcs)) {
                TRACE_DATA("qcs retrieval error", QMUX_EV_QCC_RECV, qcc->conn);
                goto err;
        }

        if (!qcs) {
                TRACE_DATA("already closed stream", QMUX_EV_QCC_RECV, qcc->conn);
                goto out;
        }

        /* RFC 9000 4.5. Stream Final Size
         *
         * Once a final size for a stream is known, it cannot change.  If a
         * RESET_STREAM or STREAM frame is received indicating a change in the
         * final size for the stream, an endpoint SHOULD respond with an error
         * of type FINAL_SIZE_ERROR; see Section 11 for details on error
         * handling.
         */
        if (((qcs->flags & QC_SF_SIZE_KNOWN) && offset + len > qcs->rx.offset_max) ||
            (fin && offset + len < qcs->rx.offset_max)) {
                TRACE_ERROR("final size error", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV|QMUX_EV_PROTO_ERR, qcc->conn, qcs);
                qcc_set_error(qcc, QC_ERR_FINAL_SIZE_ERROR, 0,
                              muxc_tevt_type_proto_err);
                goto err;
        }

        if (qcs_is_close_remote(qcs)) {
                TRACE_DATA("skipping STREAM for remotely closed", QMUX_EV_QCC_RECV, qcc->conn);
                goto out;
        }

        if (offset + len < qcs->rx.offset ||
            (offset + len == qcs->rx.offset && (!fin || (qcs->flags & QC_SF_SIZE_KNOWN)))) {
                TRACE_DATA("already received offset", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV, qcc->conn, qcs);
                goto out;
        }

        TRACE_PROTO("receiving STREAM", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV, qcc->conn, qcs);
        qcs_idle_open(qcs);

        if (offset + len > qcs->rx.offset_max) {
                uint64_t diff = offset + len - qcs->rx.offset_max;
                qcs->rx.offset_max = offset + len;
                qcc->lfctl.offsets_recv += diff;

                if (offset + len > qcs->rx.msd ||
                    qcc->lfctl.offsets_recv > qcc->lfctl.md) {
                        /* RFC 9000 4.1. Data Flow Control
                         *
                         * A receiver MUST close the connection with an error
                         * of type FLOW_CONTROL_ERROR if the sender violates
                         * the advertised connection or stream data limits
                         */
                        TRACE_ERROR("flow control error", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV|QMUX_EV_PROTO_ERR,
                                    qcc->conn, qcs);
                        qcc_set_error(qcc, QC_ERR_FLOW_CONTROL_ERROR, 0,
                                      muxc_tevt_type_proto_err);
                        goto err;
                }
        }

        TRACE_DATA("newly received offset", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV, qcc->conn, qcs);
        if (offset < qcs->rx.offset) {
                size_t diff = qcs->rx.offset - offset;

                len -= diff;
                data += diff;
                offset = qcs->rx.offset;
        }

        if (len && (qcc->flags & QC_CF_WAIT_HS)) {
                if (!(qcc->conn->flags & CO_FL_EARLY_DATA)) {
                        /* Ensure 'Early-data: 1' will be set on the request. */
                        TRACE_PROTO("received early data", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV, qcc->conn, qcs);
                        qcc->conn->flags |= CO_FL_EARLY_DATA;
                }
        }

        left = len;
        while (left) {
                struct qc_stream_rxbuf *buf;
                struct proxy *px;
                struct quic_counters *prx_counters;
                ncb_sz_t ncb_off;

                buf = qcs_get_rxbuf(qcs, offset, &len);
                if (!buf) {
                        TRACE_ERROR("rxbuf alloc failure", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV, qcc->conn, qcs);
                        qcc_set_error(qcc, QC_ERR_INTERNAL_ERROR, 0,
                                      muxc_tevt_type_internal_err);
                        goto err;
                }

                /* For oldest buffer, ncb_advance() may already have been performed. */
                ncb_off = offset - MAX(qcs->rx.offset, buf->off_node.key);

                ncb_ret = ncb_add(&buf->ncb, ncb_off, data, len, NCB_ADD_COMPARE);
                switch (ncb_ret) {
                case NCB_RET_OK:
                        break;

                case NCB_RET_DATA_REJ:
                        /* RFC 9000 2.2. Sending and Receiving Data
                         *
                         * An endpoint could receive data for a stream at the
                         * same stream offset multiple times. Data that has
                         * already been received can be discarded. The data at
                         * a given offset MUST NOT change if it is sent
                         * multiple times; an endpoint MAY treat receipt of
                         * different data at the same offset within a stream as
                         * a connection error of type PROTOCOL_VIOLATION.
                         */
                        TRACE_ERROR("overlapping data rejected", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV|QMUX_EV_PROTO_ERR,
                                    qcc->conn, qcs);
                        qcc_set_error(qcc, QC_ERR_PROTOCOL_VIOLATION, 0,
                                      muxc_tevt_type_proto_err);
                        return 1;

                case NCB_RET_GAP_SIZE:
                        TRACE_DATA("cannot bufferize frame due to gap size limit", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV,
                                   qcc->conn, qcs);
                        px = qcc->proxy;
                        prx_counters = EXTRA_COUNTERS_GET(px->extra_counters_fe, &quic_stats_module);
                        HA_ATOMIC_INC(&prx_counters->ncbuf_gap_limit);
                        return 1;
                }

                offset += len;
                data += len;
                left -= len;
                len = left;
        }

        if (fin)
                qcs->flags |= QC_SF_SIZE_KNOWN;

        if (qcs->flags & QC_SF_SIZE_KNOWN &&
            qcs->rx.offset_max == qcs->rx.offset + qcs_rx_avail_data(qcs)) {
                qcs_close_remote(qcs);
        }

        while ((qcs_rx_avail_data(qcs) && !(qcs->flags & QC_SF_DEM_FULL)) ||
               unlikely(fin_standalone && qcs_is_close_remote(qcs))) {

                ret = qcc_decode_qcs(qcc, qcs);
                LIST_DEL_INIT(&qcs->el_recv);
                qcc_refresh_timeout(qcc);

                if (ret <= 0)
                        break;

                BUG_ON_HOT(fin_standalone); /* On fin_standalone <ret> should be NULL, which ensures no infinite loop. */
        }

        /* Ensure that an idle backend conn is freed if it cannot open new
         * stream. This is only performed for QUIC which directly calls
         * qcc_recv() and is not compatible with QMux. For the latter, dead
         * connection should still be detected after recv via qcc_io_process().
         */
        if (conn_is_quic(qcc->conn) && conn_is_back(qcc->conn) && qcc_is_dead(qcc)) {
                TRACE_STATE("releasing dead connection after STREAM decoding", QMUX_EV_QCC_RECV, qcc->conn);
                qcc_release(qcc);
                return 0;
        }

        /* Wakeup tasklet if QCS inserted in purge list. May happen after remote closure. */
        if (qcs_is_completed(qcs))
                tasklet_wakeup(qcc->wait_event.tasklet);

 out:
        TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
        return 0;

 err:
        TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
        return 1;
}

/* Handle a new MAX_DATA frame. <max> must contains the maximum data field of
 * the frame.
 *
 * Returns 0 on success else non-zero.
 */
int qcc_recv_max_data(struct qcc *qcc, uint64_t max)
{
        const int blocked_soft = qfctl_sblocked(&qcc->tx.fc);
        int unblock_soft = 0, unblock_real = 0;

        TRACE_ENTER(QMUX_EV_QCC_RECV, qcc->conn);

        TRACE_PROTO("receiving MAX_DATA", QMUX_EV_QCC_RECV, qcc->conn);
        if (qfctl_set_max(&qcc->tx.fc, max, &unblock_soft, &unblock_real)) {
                TRACE_DATA("increase remote max-data", QMUX_EV_QCC_RECV, qcc->conn);

                if (unblock_real)
                        tasklet_wakeup(qcc->wait_event.tasklet);

                if (unblock_soft)
                        qcc_notify_fctl(qcc);

                /* Refresh frms list only if this would result in newer data :
                 * a. flow-control is not real blocked
                 * b. soft off was equal or greater than previous limit
                 */
                if (!qfctl_rblocked(&qcc->tx.fc) && blocked_soft)
                        qcc_clear_frms(qcc);
        }

        TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
        return 0;
}

/* Handle a new MAX_STREAM_DATA frame. <max> must contains the maximum data
 * field of the frame and <id> is the identifier of the QUIC stream.
 *
 * Returns 0 on success else non-zero. On error, the received frame should not
 * be acknowledged.
 */
int qcc_recv_max_stream_data(struct qcc *qcc, uint64_t id, uint64_t max)
{
        struct qcs *qcs;

        TRACE_ENTER(QMUX_EV_QCC_RECV, qcc->conn);

        if (qcc->flags & QC_CF_ERRL) {
                TRACE_DATA("connection on error", QMUX_EV_QCC_RECV, qcc->conn);
                goto err;
        }

        /* RFC 9000 19.10. MAX_STREAM_DATA Frames
         *
         * Receiving a MAX_STREAM_DATA frame for a locally
         * initiated stream that has not yet been created MUST be treated as a
         * connection error of type STREAM_STATE_ERROR.  An endpoint that
         * receives a MAX_STREAM_DATA frame for a receive-only stream MUST
         * terminate the connection with error STREAM_STATE_ERROR.
         */
        if (qcc_get_qcs(qcc, id, 0, 1, &qcs))
                goto err;

        if (qcs) {
                const int blocked_soft = qfctl_sblocked(&qcs->tx.fc);
                int unblock_soft = 0, unblock_real = 0;

                TRACE_PROTO("receiving MAX_STREAM_DATA", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV, qcc->conn, qcs);
                if (qfctl_set_max(&qcs->tx.fc, max, &unblock_soft, &unblock_real)) {
                        TRACE_DATA("increase remote max-stream-data", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV, qcc->conn, qcs);
                        if (unblock_real) {
                                /* TODO optim: only wakeup IO-CB if stream has data to sent. */
                                tasklet_wakeup(qcc->wait_event.tasklet);
                        }

                        if (unblock_soft) {
                                tot_time_stop(&qcs->timer.fctl);
                                qcs_notify_send(qcs);
                        }

                        /* Same refresh condition as qcc_recv_max_data(). */
                        if (!qfctl_rblocked(&qcs->tx.fc) && blocked_soft)
                                qcc_clear_frms(qcc);
                }
        }

        if (qcc_may_expire(qcc) && !qcc->nb_hreq)
                qcc_refresh_timeout(qcc);

        TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
        return 0;

 err:
        TRACE_DEVEL("leaving on error", QMUX_EV_QCC_RECV, qcc->conn);
        return 1;
}

/* Handle a MAX_STREAMS frame. <max> must contains the cumulative number of
 * streams that can be opened. <bidi> is a boolean set if this refers to
 * bidirectional streams.
 *
 * Returns 0 on success else non-zero. On error, the received frame should not
 * be acknowledged.
 */
int qcc_recv_max_streams(struct qcc *qcc, uint64_t max, int bidi)
{
        TRACE_ENTER(QMUX_EV_QCC_RECV, qcc->conn);

        if (qcc->flags & QC_CF_ERRL) {
                TRACE_DATA("connection on error", QMUX_EV_QCC_RECV, qcc->conn);
                goto err;
        }

        /* RFC 9000 19.11. MAX_STREAMS Frames
         *
         *  This value cannot exceed 2^60, as it is not possible to
         *  encode stream IDs larger than 2^62-1. Receipt of a frame that
         *  permits opening of a stream larger than this limit MUST be treated
         *  as a connection error of type FRAME_ENCODING_ERROR.
         */
        if (max > QUIC_VARINT_8_BYTE_MAX) {
                TRACE_ERROR("invalid MAX_STREAMS value", QMUX_EV_QCC_RECV, qcc->conn);
                qcc_set_error(qcc, QC_ERR_FRAME_ENCODING_ERROR, 0,
                              muxc_tevt_type_proto_err);
                goto err;
        }

        TRACE_PROTO("receiving MAX_STREAMS", QMUX_EV_QCC_RECV, qcc->conn);
        if (bidi) {
                if (max > qcc->rfctl.ms_bidi) {
                        TRACE_DATA("increase remote max-streams-bidi", QMUX_EV_QCC_RECV, qcc->conn);
                        qcc->rfctl.ms_bidi = max;
                }
        }
        else {
                /* TODO no extra unidirectional streams open after connection
                 * startup, so uni MAX_STREAMS flow-control is not necessary
                 * for now.
                 */
        }

        TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
        return 0;

 err:
        TRACE_DEVEL("leaving on error", QMUX_EV_QCC_RECV, qcc->conn);
        return 1;
}

/* Handle a new RESET_STREAM frame from stream ID <id> with error code <err>
 * and final stream size <final_size>.
 *
 * Returns 0 on success else non-zero. On error, the received frame should not
 * be acknowledged.
 */
int qcc_recv_reset_stream(struct qcc *qcc, uint64_t id, uint64_t err, uint64_t final_size)
{
        struct qcs *qcs;
        struct qc_stream_rxbuf *b;
        int prev_glitches = qcc->glitches;
        uint64_t to_consume;

        TRACE_ENTER(QMUX_EV_QCC_RECV, qcc->conn);

        if (qcc->flags & QC_CF_ERRL) {
                TRACE_DATA("connection on error", QMUX_EV_QCC_RECV, qcc->conn);
                goto err;
        }

        /* RFC 9000 19.4. RESET_STREAM Frames
         *
         * An endpoint that receives a RESET_STREAM frame for a send-only stream
         * MUST terminate the connection with error STREAM_STATE_ERROR.
         */
        if (qcc_get_qcs(qcc, id, 1, 0, &qcs)) {
                TRACE_ERROR("RESET_STREAM for send-only stream received", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV, qcc->conn, qcs);
                goto err;
        }

        /* RFC 9000 3.2. Receiving Stream States
         *
         * A RESET_STREAM signal might be suppressed or withheld
         * if stream data is completely received and is buffered to be read by
         * the application. If the RESET_STREAM is suppressed, the receiving
         * part of the stream remains in "Data Recvd".
         */
        if (!qcs || qcs_is_close_remote(qcs))
                goto out;

        TRACE_PROTO("receiving RESET_STREAM", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV, qcc->conn, qcs);
        qcs_idle_open(qcs);

        /* Ensure stream closure is not forbidden by application protocol. */
        if (qcc->app_ops->close) {
                if (qcc->app_ops->close(qcs, QCC_APP_OPS_CLOSE_SIDE_RD)) {
                        TRACE_ERROR("closure rejected by app layer", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV, qcc->conn, qcs);
                        goto out;
                }
        }

        if (qcs->rx.offset_max > final_size ||
            ((qcs->flags & QC_SF_SIZE_KNOWN) && qcs->rx.offset_max != final_size)) {
                TRACE_ERROR("final size error on RESET_STREAM",
                            QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV|QMUX_EV_PROTO_ERR, qcc->conn, qcs);
                qcc_set_error(qcc, QC_ERR_FINAL_SIZE_ERROR, 0,
                              muxc_tevt_type_proto_err);
                goto err;
        }

        /* Discarded data to account for in connection flow control. */
        to_consume = final_size - qcs->rx.offset;

        /* Ensure final size respects the current stream and connection flow control limits. */
        if (final_size > qcs->rx.msd ||
            qcc->lfctl.offsets_recv + to_consume > qcc->lfctl.md) {
                TRACE_ERROR("flow control violation on RESET_STREAM final size",
                            QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV|QMUX_EV_PROTO_ERR, qcc->conn, qcs);
                qcc_set_error(qcc, QC_ERR_FLOW_CONTROL_ERROR, 0,
                              muxc_tevt_type_proto_err);
                goto err;
        }

        qcc->lfctl.offsets_recv += to_consume;

        qcs->flags |= QC_SF_SIZE_KNOWN|QC_SF_RECV_RESET;
        qcs_close_remote(qcs);
        /* Wakeup tasklet if QCS inserted in purge list. May happen after remote closure. */
        if (qcs_is_completed(qcs))
                tasklet_wakeup(qcc->wait_event.tasklet);

        /* RFC 9000 3.2. Receiving Stream States
         *
         * An
         * implementation MAY interrupt delivery of stream data, discard any
         * data that was not consumed, and signal the receipt of the
         * RESET_STREAM.
         */
        while (!eb_is_empty(&qcs->rx.bufs)) {
                b = container_of(eb64_first(&qcs->rx.bufs),
                                 struct qc_stream_rxbuf, off_node);
                qcs_free_rxbuf(qcs, b);
        }

        qcc_account_for_conn_fctl(qcc, to_consume);
        /* Ensure final size checks coherence for future STREAM frames. */
        qcs->rx.offset_max = final_size;

        /* Remove stream from recv_list if present. */
        LIST_DEL_INIT(&qcs->el_recv);

        /* No more data will be received after peer reset. QCS can be purged
         * if there is nothing to sent on our side.
         */
        if (!qcs_sc(qcs) && !LIST_INLIST(&qcs->el_send)) {
                qcs_close_local(qcs);
                TRACE_STATE("add stream in purg_list", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV, qcc->conn, qcs);
                LIST_APPEND(&qcs->qcc->purg_list, &qcs->el_send);
                tasklet_wakeup(qcs->qcc->wait_event.tasklet);
        }
        else {
                /* Wake-up upper stream layer to notify about reset on read. */
                qcs_alert(qcs);
        }

 out:
        if (qcc->glitches != prev_glitches && !(qcc->flags & QC_CF_IS_BACK))
                session_add_glitch_ctr(qcc->conn->owner, qcc->glitches - prev_glitches);

        TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
        return 0;

 err:
        TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
        return 1;
}

/* Handle a new STOP_SENDING frame for stream ID <id>. The error code should be
 * specified in <err>.
 *
 * Returns 0 on success else non-zero. On error, the received frame should not
 * be acknowledged.
 */
int qcc_recv_stop_sending(struct qcc *qcc, uint64_t id, uint64_t err)
{
        struct qcs *qcs;
        int prev_glitches = qcc->glitches;

        TRACE_ENTER(QMUX_EV_QCC_RECV, qcc->conn);

        if (qcc->flags & QC_CF_ERRL) {
                TRACE_DATA("connection on error", QMUX_EV_QCC_RECV, qcc->conn);
                goto err;
        }

        /* RFC 9000 19.5. STOP_SENDING Frames
         *
         * Receiving a STOP_SENDING frame for a
         * locally initiated stream that has not yet been created MUST be
         * treated as a connection error of type STREAM_STATE_ERROR.  An
         * endpoint that receives a STOP_SENDING frame for a receive-only stream
         * MUST terminate the connection with error STREAM_STATE_ERROR.
         */
        if (qcc_get_qcs(qcc, id, 0, 1, &qcs))
                goto err;

        if (!qcs)
                goto out;

        TRACE_PROTO("receiving STOP_SENDING", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV, qcc->conn, qcs);

        /* RFC 9000 3.5. Solicited State Transitions
         *
         * An endpoint is expected to send another STOP_SENDING frame if a
         * packet containing a previous STOP_SENDING is lost.  However, once
         * either all stream data or a RESET_STREAM frame has been received for
         * the stream -- that is, the stream is in any state other than "Recv"
         * or "Size Known" -- sending a STOP_SENDING frame is unnecessary.
         */

        /* TODO thanks to previous RFC clause, STOP_SENDING is ignored if current stream
         * has already been closed locally. This is useful to not emit multiple
         * RESET_STREAM for a single stream. This is functional if stream is
         * locally closed due to all data transmitted, but in this case the RFC
         * advises to use an explicit RESET_STREAM.
         */
        if (qcs_is_close_local(qcs)) {
                TRACE_STATE("ignoring STOP_SENDING", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV, qcc->conn, qcs);
                goto out;
        }

        qcs_idle_open(qcs);

        if (qcc->app_ops->close) {
                if (qcc->app_ops->close(qcs, QCC_APP_OPS_CLOSE_SIDE_WR)) {
                        TRACE_ERROR("closure rejected by app layer", QMUX_EV_QCC_RECV|QMUX_EV_QCS_RECV, qcc->conn, qcs);
                        goto out;
                }
        }

        if (qcs->sd) {
                /* Manually set EOS if FIN already reached as futures RESET_STREAM will be ignored in this case. */
                if (qcs_sc(qcs) && se_fl_test(qcs->sd, SE_FL_EOI)) {
                        se_fl_set(qcs->sd, SE_FL_EOS);
                        se_report_term_evt(qcs->sd, (qcc->flags & QC_CF_ERR_CONN ? se_tevt_type_rcv_err : se_tevt_type_eos));
                        qcs_alert(qcs);
                }
                else {
                        se_report_term_evt(qcs->sd, se_tevt_type_rst_rcvd);
                }

                /* If not defined yet, set abort info for the sedesc */
                if (!qcs->sd->abort_info.info) {
                        qcs->sd->abort_info.info = (SE_ABRT_SRC_MUX_QUIC << SE_ABRT_SRC_SHIFT);
                        qcs->sd->abort_info.code = err;
                }
        }

        /* RFC 9000 3.5. Solicited State Transitions
         *
         * An endpoint that receives a STOP_SENDING frame
         * MUST send a RESET_STREAM frame if the stream is in the "Ready" or
         * "Send" state.  If the stream is in the "Data Sent" state, the
         * endpoint MAY defer sending the RESET_STREAM frame until the packets
         * containing outstanding data are acknowledged or declared lost.  If
         * any outstanding data is declared lost, the endpoint SHOULD send a
         * RESET_STREAM frame instead of retransmitting the data.
         *
         * An endpoint SHOULD copy the error code from the STOP_SENDING frame to
         * the RESET_STREAM frame it sends, but it can use any application error
         * code.
         */
        qcc_reset_stream(qcs, err, 0);

        if (qcc_may_expire(qcc) && !qcc->nb_hreq)
                qcc_refresh_timeout(qcc);

 out:
        if (qcc->glitches != prev_glitches && !(qcc->flags & QC_CF_IS_BACK))
                session_add_glitch_ctr(qcc->conn->owner, qcc->glitches - prev_glitches);

        TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
        return 0;

 err:
        TRACE_DEVEL("leaving on error", QMUX_EV_QCC_RECV, qcc->conn);
        return 1;
}

/* Signal the closing of remote stream with id <id>. Flow-control for new
 * streams may be allocated for the peer if needed.
 */
static int qcc_release_remote_stream(struct qcc *qcc, uint64_t id)
{
        struct quic_frame *frm;
        uint64_t conn_max, rem, non_extra, inc;

        TRACE_ENTER(QMUX_EV_QCS_END, qcc->conn);

        if (quic_stream_is_bidi(id)) {
                const uint64_t max = qcc_max_strm_bidi_remote(qcc->conn);
                /* The peer must not have been authorized to open a stream outside of this range. */
                BUG_ON(qcc->lfctl.ms_bidi > max);
                if (qcc->lfctl.ms_bidi == max) {
                        TRACE_DATA("maximum streams value reached", QMUX_EV_QCC_SEND, qcc->conn);
                        goto out;
                }

                ++qcc->lfctl.cl_bidi_r;
                /* MAX_STREAMS needed if closed streams value more than twice
                 * the initial window or reaching the stream ID limit.
                 */
                if (qcc->lfctl.cl_bidi_r > qcc->lfctl.ms_bidi_rel / 2 ||
                    qcc->lfctl.cl_bidi_r + qcc->lfctl.ms_bidi == max) {

                        BUG_ON(qcc->lfctl.ms_bidi_rel < qcc->lfctl.cl_bidi_r);
                        rem = qcc->lfctl.ms_bidi_rel - qcc->lfctl.cl_bidi_r;
                        /* if every streams are closed, decrement extra stream accounting by 1 */
                        non_extra = !rem ? 1 : 0;

                        if (!(qcc->flags & QC_CF_IS_BACK) && global.tune.streams_elasticity) {
                                /* If stream elasticity is active, first decrement closed from extra streams. */
                                if (qcc->lfctl.ms_bidi_rel > 1) {
                                        _HA_ATOMIC_SUB(&tg_ctx->committed_extra_streams,
                                                       qcc->lfctl.cl_bidi_r - non_extra);
                                }

                                /* Now calculate the available streams. */
                                conn_max = conn_calc_max_streams(qcc->lfctl.ms_bidi_init);
                                if (conn_max <= rem) {
                                        /* More streams already consumed than currently allowed,
                                         * keep the current flow control limit.
                                         */
                                        qcc->lfctl.ms_bidi_rel = rem;
                                        qcc->lfctl.cl_bidi_r = 0;
                                        goto out;
                                }

                                /* Update flow control limit up to the allowed elasticity limit. */
                                inc = conn_max - rem;
                                _HA_ATOMIC_ADD(&tg_ctx->committed_extra_streams, inc - non_extra);
                                qcc->lfctl.ms_bidi_rel = rem + inc;
                        }
                        else {
                                /* Stream elasticity not active, flow control increase remains static. */
                                inc = qcc->lfctl.cl_bidi_r;
                        }

                        TRACE_DATA("increase max stream limit with MAX_STREAMS_BIDI", QMUX_EV_QCC_SEND, qcc->conn);
                        frm = qc_frm_alloc(QUIC_FT_MAX_STREAMS_BIDI);
                        if (!frm) {
                                qcc_set_error(qcc, QC_ERR_INTERNAL_ERROR, 0,
                                              muxc_tevt_type_internal_err);
                                goto err;
                        }

                        frm->max_streams_bidi.max_streams = qcc->lfctl.ms_bidi + inc;
                        LIST_APPEND(&qcc->lfctl.frms, &frm->list);
                        tasklet_wakeup(qcc->wait_event.tasklet);

                        qcc->lfctl.ms_bidi += inc;
                        qcc->lfctl.cl_bidi_r = 0;
                }
        }
        else {
                /* TODO unidirectional stream flow control with MAX_STREAMS_UNI
                 * emission not implemented. It should be unnecessary for
                 * HTTP/3 but may be required if other application protocols
                 * are supported.
                 */
        }

 out:
        TRACE_LEAVE(QMUX_EV_QCS_END, qcc->conn);
        return 0;

 err:
        TRACE_DEVEL("leaving on error", QMUX_EV_QCS_END, qcc->conn);
        return 1;
}

/* detaches the QUIC stream from its QCC and releases it to the QCS pool. */
static void qcs_destroy(struct qcs *qcs)
{
        struct qcc *qcc = qcs->qcc;
        struct connection *conn = qcc->conn;
        const uint64_t id = qcs->id;

        TRACE_ENTER(QMUX_EV_QCS_END, conn, qcs);

        if (!(qcc->flags & (QC_CF_ERR_CONN|QC_CF_ERRL))) {
                /* MUST not removed a stream with sending prepared data left. This is
                 * to ensure consistency on connection flow-control calculation.
                 */
                BUG_ON(qcs->tx.fc.off_soft != qcs->tx.fc.off_real);

                if (quic_stream_is_remote(qcc, id))
                        qcc_release_remote_stream(qcc, id);
        }

        /* If some data are not yet consumed, account them in conn flow
         * control. On frontend, this may happen when FE stream is released
         * prior to request completion.
         *
         * TODO QCS should not be released if not closed remotely to guarantee
         * the connection flow control consistency
         */
        if (!qcs_is_close_remote(qcs) && qcs->rx.offset_max > qcs->rx.offset) {
                COUNT_IF(1, "QCS is released while not remotely closed. "
                            "This may cause miscalculation in conn flow control.");
                qcc_account_for_conn_fctl(qcc, qcs->rx.offset_max - qcs->rx.offset);
        }

        qcs_free(qcs);

        /* Rearm http-keep-alive timeout when last request stream is freed. */
        if (!conn_is_back(qcc->conn) && qcc_may_expire(qcc) && !qcc->nb_hreq) {
                qcc_reset_idle_start(qcc);
                qcc_refresh_timeout(qcc);
        }

        TRACE_LEAVE(QMUX_EV_QCS_END, conn);
}

/* Prepare a STREAM frame for <qcs> instance using <out> as payload. The frame
 * is appended in <frm_list>. Set <fin> if this is supposed to be the last
 * stream frame. If <out> is NULL an empty STREAM frame is built : this may be
 * useful if FIN needs to be sent without any data left. Frame length will be
 * truncated if greater than <fc_conn_wnd>. This allows to prepare several
 * frames in a loop while respecting connection flow control window.
 *
 * Returns the payload length of the STREAM frame or a negative error code.
 */
static int qcs_build_stream_frm(struct qcs *qcs, struct buffer *out, char fin,
                                struct list *frm_list, uint64_t window_conn)
{
        struct qcc *qcc = qcs->qcc;
        struct quic_frame *frm;
        const uint64_t window_stream = qfctl_rcap(&qcs->tx.fc);
        const uint64_t bytes = qcs_prep_bytes(qcs);
        uint64_t total;

        TRACE_ENTER(QMUX_EV_QCS_SEND, qcc->conn, qcs);

        /* This must only be called if there is data left, or at least a standalone FIN. */
        BUG_ON((!out || !b_data(out)) && !fin);

        total = bytes;

        /* do not exceed stream flow control limit */
        if (total > window_stream) {
                TRACE_DATA("do not exceed stream flow control", QMUX_EV_QCS_SEND, qcc->conn, qcs);
                total = window_stream;
        }

        /* do not exceed connection flow control limit */
        if (total > window_conn) {
                TRACE_DATA("do not exceed conn flow control", QMUX_EV_QCS_SEND, qcc->conn, qcs);
                total = window_conn;
        }

        /* Reset FIN if bytes to send is capped by flow control. */
        if (total < bytes)
                fin = 0;

        if (!total && !fin) {
                /* No need to send anything if total is NULL and no FIN to signal. */
                TRACE_LEAVE(QMUX_EV_QCS_SEND, qcc->conn, qcs);
                return 0;
        }

        TRACE_PROTO("sending STREAM frame", QMUX_EV_QCS_SEND, qcc->conn, qcs);
        frm = qc_frm_alloc(QUIC_FT_STREAM_8);
        if (!frm) {
                TRACE_ERROR("frame alloc failure", QMUX_EV_QCS_SEND, qcc->conn, qcs);
                goto err;
        }

        frm->stream.stream =
          conn_is_quic(qcc->conn) ? (void *)qcs->tx.stream : (void *)qcs;
        frm->stream.id = qcs->id;
        frm->stream.offset = 0;
        frm->stream.dup = 0;

        if (total) {
                frm->stream.buf = out;
                frm->stream.data = (unsigned char *)b_peek(out, b_data(out) - bytes);
        }
        else {
                /* Empty STREAM frame. */
                frm->stream.buf = NULL;
                frm->stream.data = NULL;
        }

        /* FIN is positioned only when the buffer has been totally emptied. */
        if (fin)
                frm->type |= QUIC_STREAM_FRAME_TYPE_FIN_BIT;

        if (qcs->tx.fc.off_real) {
                frm->type |= QUIC_STREAM_FRAME_TYPE_OFF_BIT;
                frm->stream.offset = qcs->tx.fc.off_real;
        }

        /* Always set length bit as we do not know if there is remaining frames
         * in the final packet after this STREAM.
         */
        frm->type |= QUIC_STREAM_FRAME_TYPE_LEN_BIT;
        frm->stream.len = total;

        LIST_APPEND(frm_list, &frm->list);

 out:
        {
                struct qcs_build_stream_trace_arg arg = {
                        .len = frm->stream.len, .fin = fin,
                        .offset = frm->stream.offset,
                };
                TRACE_LEAVE(QMUX_EV_QCS_SEND|QMUX_EV_QCS_BUILD_STRM,
                            qcc->conn, qcs, &arg);
        }

        return total;

 err:
        TRACE_LEAVE(QMUX_EV_QCS_SEND, qcc->conn, qcs);
        return -1;
}

/* Returns true if subscribe set, false otherwise. */
static int qcc_subscribe_send(struct qcc *qcc)
{
        struct connection *conn = qcc->conn;

        /* Do not subscribe if lower layer in error. */
        if (conn->flags & CO_FL_ERROR)
                return 0;

        if (qcc->wait_event.events & SUB_RETRY_SEND)
                return 1;

        TRACE_DEVEL("subscribe for send", QMUX_EV_QCC_SEND, qcc->conn);
        conn->xprt->subscribe(conn, conn->xprt_ctx, SUB_RETRY_SEND, &qcc->wait_event);
        return 1;
}

static int qcc_quic_send_frames(struct qcc *qcc, struct list *frms, int stream)
{
        enum quic_tx_err ret;
        struct quic_pacer *pacer = NULL;

        TRACE_ENTER(QMUX_EV_QCC_SEND, qcc->conn);

        if (stream && qcc_is_pacing_active(qcc->conn))
                pacer = &qcc->tx.pacer;

        ret = qc_send_mux(qcc->conn->handle.qc, frms, pacer);
        if (ret == QUIC_TX_ERR_FATAL) {
                TRACE_DEVEL("error on sending", QMUX_EV_QCC_SEND, qcc->conn);
                qcc_subscribe_send(qcc);
                goto err;
        }

        /* If there is frames left at this stage, transport layer is blocked.
         * Subscribe on it to retry later.
         */
        if (!LIST_ISEMPTY(frms) && ret != QUIC_TX_ERR_PACING) {
                TRACE_DEVEL("remaining frames to send", QMUX_EV_QCC_SEND, qcc->conn);
                qcc_subscribe_send(qcc);
                goto err;
        }

        TRACE_LEAVE(QMUX_EV_QCC_SEND, qcc->conn);
        return ret == QUIC_TX_ERR_PACING ? 1 : 0;

 err:
        TRACE_DEVEL("leaving on error", QMUX_EV_QCC_SEND, qcc->conn);
        return -1;
}

/* Wrapper for send on transport layer. Send a list of frames <frms> for the
 * connection <qcc>.
 *
 * Returns 0 if all data sent with success. On fatal error, a negative error
 * code is returned. A positive 1 is used if emission should be paced.
 */
static int qcc_send_frames(struct qcc *qcc, struct list *frms, int stream)
{
        if (LIST_ISEMPTY(frms)) {
                TRACE_DEVEL("leaving on no frame to send", QMUX_EV_QCC_SEND, qcc->conn);
                return -1;
        }

        return conn_is_quic(qcc->conn) ? qcc_quic_send_frames(qcc, frms, stream) :
                                         qcc_qmux_send_frames(qcc, frms);
}

/* Emit a RESET_STREAM on <qcs>.
 *
 * Returns 0 if the frame has been successfully sent else non-zero.
 */
static int qcs_send_reset(struct qcs *qcs)
{
        struct list frms = LIST_HEAD_INIT(frms);
        struct quic_frame *frm;

        TRACE_ENTER(QMUX_EV_QCS_SEND, qcs->qcc->conn, qcs);

        frm = qc_frm_alloc(QUIC_FT_RESET_STREAM);
        if (!frm) {
                TRACE_LEAVE(QMUX_EV_QCS_SEND, qcs->qcc->conn, qcs);
                return 1;
        }

        frm->reset_stream.id = qcs->id;
        frm->reset_stream.app_error_code = qcs->rs_err;
        frm->reset_stream.final_size = qcs->tx.fc.off_real;

        LIST_APPEND(&frms, &frm->list);
        if (qcc_send_frames(qcs->qcc, &frms, 0)) {
                if (!LIST_ISEMPTY(&frms))
                        qc_frm_free(conn_is_quic(qcs->qcc->conn) ? qcs->qcc->conn->handle.qc : NULL, &frm);
                TRACE_DEVEL("cannot send RESET_STREAM", QMUX_EV_QCS_SEND, qcs->qcc->conn, qcs);
                return 1;
        }

        qcs_close_local(qcs);
        qcs->flags &= ~QC_SF_TO_RESET;

        TRACE_LEAVE(QMUX_EV_QCS_SEND, qcs->qcc->conn, qcs);
        return 0;
}

/* Emit a STOP_SENDING on <qcs>.
 *
 * Returns 0 if the frame has been successfully sent else non-zero.
 */
static int qcs_send_stop_sending(struct qcs *qcs)
{
        struct list frms = LIST_HEAD_INIT(frms);
        struct quic_frame *frm;
        struct qcc *qcc = qcs->qcc;

        TRACE_ENTER(QMUX_EV_QCS_SEND, qcs->qcc->conn, qcs);

        /* RFC 9000 3.3. Permitted Frame Types
         *
         * A
         * receiver MAY send a STOP_SENDING frame in any state where it has not
         * received a RESET_STREAM frame -- that is, states other than "Reset
         * Recvd" or "Reset Read". However, there is little value in sending a
         * STOP_SENDING frame in the "Data Recvd" state, as all stream data has
         * been received. A sender could receive either of these two types of
         * frames in any state as a result of delayed delivery of packets.¶
         */
        if (qcs_is_close_remote(qcs)) {
                TRACE_STATE("skip STOP_SENDING on remote already closed", QMUX_EV_QCS_SEND, qcc->conn, qcs);
                goto done;
        }

        frm = qc_frm_alloc(QUIC_FT_STOP_SENDING);
        if (!frm) {
                TRACE_LEAVE(QMUX_EV_QCS_SEND, qcs->qcc->conn, qcs);
                return 1;
        }

        frm->stop_sending.id = qcs->id;
        frm->stop_sending.app_error_code = qcs->ss_err;

        LIST_APPEND(&frms, &frm->list);
        if (qcc_send_frames(qcs->qcc, &frms, 0)) {
                if (!LIST_ISEMPTY(&frms))
                        qc_frm_free(conn_is_quic(qcc->conn) ? qcc->conn->handle.qc : NULL, &frm);
                TRACE_DEVEL("cannot send STOP_SENDING", QMUX_EV_QCS_SEND, qcs->qcc->conn, qcs);
                return 1;
        }

 done:
        qcs->flags &= ~QC_SF_TO_STOP_SENDING;

        TRACE_LEAVE(QMUX_EV_QCS_SEND, qcs->qcc->conn, qcs);
        return 0;
}

/* Used internally by qcc_io_send function. Proceed to send for <qcs>. A STREAM
 * frame is generated pointing to QCS stream descriptor content and inserted in
 * <frms> list. Frame length will be truncated if greater than <window_conn>.
 * This allows to prepare several frames in a loop while respecting connection
 * flow control window.
 *
 * Returns the payload length of the STREAM frame or a negative error code.
 */
static int qcs_send(struct qcs *qcs, struct list *frms, uint64_t window_conn)
{
        struct qcc *qcc = qcs->qcc;
        struct buffer *out = qcs_tx_buf(qcs);
        int flen = 0;
        const char fin = qcs->flags & QC_SF_FIN_STREAM;

        TRACE_ENTER(QMUX_EV_QCS_SEND, qcc->conn, qcs);

        /* Cannot send STREAM on remote unidirectional streams. */
        BUG_ON(quic_stream_is_uni(qcs->id) && quic_stream_is_remote(qcc, qcs->id));

        /* This function must not be called if there is nothing to send. */
        BUG_ON(!fin && !qcs_prep_bytes(qcs));

        /* Skip STREAM frame allocation if already subscribed for send.
         * Happens on sendto transient error or network congestion.
         */
        if (qcc->wait_event.events & SUB_RETRY_SEND) {
                TRACE_DEVEL("already subscribed for sending",
                            QMUX_EV_QCS_SEND, qcc->conn, qcs);
                goto err;
        }

        /* Build a new STREAM frame with <out> buffer. */
        flen = qcs_build_stream_frm(qcs, out, fin, frms, window_conn);
        if (flen < 0)
                goto err;

 out:
        TRACE_LEAVE(QMUX_EV_QCS_SEND, qcc->conn, qcs);
        return flen;

 err:
        TRACE_DEVEL("leaving on error", QMUX_EV_QCS_SEND, qcc->conn, qcs);
        return -1;
}

/* Emit prepared flow-control related frames from <qcc> list.
 *
 * Returns 0 on success else non zero.
 */
static int qcc_emit_fctl(struct qcc *qcc)
{
        struct list single_frm = LIST_HEAD_INIT(single_frm);
        struct quic_frame *frm;
        struct eb64_node *node;
        struct qcs *qcs;
        int64_t id;

        TRACE_ENTER(QMUX_EV_QCC_END, qcc->conn);

        while (!LIST_ISEMPTY(&qcc->lfctl.frms)) {
                /* Each frame is sent individually in a single element list. */
                frm = LIST_ELEM(qcc->lfctl.frms.n, struct quic_frame *, list);
                id = frm->type == QUIC_FT_MAX_STREAM_DATA ?
                     frm->max_stream_data.id : -1;

                LIST_DELETE(&frm->list);
                LIST_APPEND(&single_frm, &frm->list);

                if (qcc_send_frames(qcc, &single_frm, 0)) {
                        if (!LIST_ISEMPTY(&single_frm)) {
                                /* replace frame that cannot be sent in qcc list. */
                                LIST_DELETE(&frm->list);
                                LIST_INSERT(&qcc->lfctl.frms, &frm->list);
                        }
                        goto err;
                }

                BUG_ON_HOT(!LIST_ISEMPTY(&single_frm));

                /* If frame is MAX_STREAM_DATA, reset corresponding QCS msd_frm pointer. */
                if (id >= 0) {
                        node = eb64_lookup(&qcc->streams_by_id, id);
                        if (node) {
                                qcs = eb64_entry(node, struct qcs, by_id);
                                qcs->tx.msd_frm = NULL;
                        }
                }
        }

        TRACE_LEAVE(QMUX_EV_QCC_SEND, qcc->conn);
        return 0;

 err:
        TRACE_DEVEL("leaving on error", QMUX_EV_QCC_SEND, qcc->conn);
        return 1;
}

/* Send RESET_STREAM/STOP_SENDING for streams in <qcc> send_list if requested.
 * Each frame is encoded and emitted separately for now. If a frame cannot be
 * sent, send_list looping is interrupted.
 *
 * Returns 0 on success else non-zero.
 */
static int qcc_emit_rs_ss(struct qcc *qcc)
{
        struct qcs *qcs, *qcs_tmp;

        TRACE_ENTER(QMUX_EV_QCC_END, qcc->conn);

        list_for_each_entry_safe(qcs, qcs_tmp, &qcc->send_list, el_send) {
                /* Stream must not be present in send_list if it has nothing to send. */
                BUG_ON(!(qcs->flags & (QC_SF_FIN_STREAM|QC_SF_TO_STOP_SENDING|QC_SF_TO_RESET)) &&
                       ((conn_is_quic(qcc->conn) && !qcs->tx.stream) || !qcs_prep_bytes(qcs)));

                /* Interrupt looping for the first stream where no RS nor SS is
                 * necessary and is not use for "metadata" transfer. These
                 * streams are always in front of the send_list.
                 */
                if (!(qcs->flags & (QC_SF_TO_STOP_SENDING|QC_SF_TO_RESET|QC_SF_TXBUB_OOB)))
                        break;

                TRACE_DATA("prepare for RS/SS transfer", QMUX_EV_QCC_SEND, qcc->conn, qcs);

                /* Each RS and SS frame is sent individually. Necessary to
                 * ensure it has been emitted as there is no transport callback
                 * for now.
                 *
                 * TODO multiplex frames to optimize sending. However, it may
                 * not be advisable to mix different streams in the same dgram
                 * to avoid interdependency in case of loss.
                 */

                if (qcs->flags & QC_SF_TO_STOP_SENDING) {
                        if (qcs_send_stop_sending(qcs))
                                goto err;

                        /* Remove stream from send_list if only SS was necessary. */
                        if (!(qcs->flags & (QC_SF_FIN_STREAM|QC_SF_TO_RESET)) &&
                            ((conn_is_quic(qcc->conn) && !qcs->tx.stream) || !qcs_prep_bytes(qcs))) {
                                LIST_DEL_INIT(&qcs->el_send);

                                if (qcs_is_completed(qcs)) {
                                        TRACE_STATE("add stream in purg_list", QMUX_EV_QCC_SEND|QMUX_EV_QCS_SEND, qcc->conn, qcs);
                                        LIST_DEL_INIT(&qcs->el_send);
                                        LIST_APPEND(&qcc->purg_list, &qcs->el_send);
                                }
                                continue;
                        }
                }

                if (qcs->flags & QC_SF_TO_RESET) {
                        if (qcs_send_reset(qcs))
                                goto err;

                        /* RFC 9000 3.3. Permitted Frame Types
                         *
                         * A sender MUST NOT send
                         * a STREAM or STREAM_DATA_BLOCKED frame for a stream in the
                         * "Reset Sent" state or any terminal state -- that is, after
                         * sending a RESET_STREAM frame.
                         */
                        LIST_DEL_INIT(&qcs->el_send);
                        if (qcs_is_completed(qcs)) {
                                TRACE_STATE("add stream in purg_list", QMUX_EV_QCC_SEND|QMUX_EV_QCS_SEND, qcc->conn, qcs);
                                LIST_APPEND(&qcc->purg_list, &qcs->el_send);
                        }
                        continue;
                }
        }

        TRACE_LEAVE(QMUX_EV_QCC_SEND, qcc->conn);
        return 0;

 err:
        TRACE_DEVEL("leaving on error", QMUX_EV_QCC_SEND, qcc->conn);
        return 1;
}

/* Encode STREAM frames into <qcc> tx frms for streams registered into
 * send_list. On each error, related stream is removed from send_list and
 * inserted into <qcs_failed> list.
 *
 * This functions also serves to emit RESET_STREAM and STOP_SENDING frames. In
 * this case, frame is emitted immediately without using <qcc> tx frms. If an
 * error occurred during this step, this is considered as fatal. Tx frms is
 * cleared and 0 is returned.
 *
 * Returns the sum of encoded payload STREAM frames length. Note that 0 can be
 * returned either if no frame was built or only empty payload frames were
 * encoded.
 */
static int qcc_build_frms(struct qcc *qcc, struct list *qcs_failed)
{
        struct list *frms = &qcc->tx.frms;
        struct qcs *qcs, *qcs_tmp, *first_qcs = NULL;
        uint64_t window_conn = qfctl_rcap(&qcc->tx.fc);
        int ret = 0, total = 0;

        TRACE_ENTER(QMUX_EV_QCC_END, qcc->conn);

        /* Frames list must first be cleared via qcc_clear_frms(). */
        BUG_ON(!LIST_ISEMPTY(&qcc->tx.frms));

        list_for_each_entry_safe(qcs, qcs_tmp, &qcc->send_list, el_send) {
                /* Check if all QCS were processed. */
                if (qcs == first_qcs)
                        break;

                TRACE_DATA("prepare for data transfer", QMUX_EV_QCC_SEND, qcc->conn, qcs);

                /* Streams with RS/SS must be handled via qcc_emit_rs_ss(). */
                BUG_ON(qcs->flags & (QC_SF_TO_STOP_SENDING|QC_SF_TO_RESET));
                /* Stream must not be present in send_list if it has nothing to send. */
                BUG_ON(!(qcs->flags & QC_SF_FIN_STREAM) && ((conn_is_quic(qcc->conn) && !qcs->tx.stream) || !qcs_prep_bytes(qcs)));

                /* Total sent bytes must not exceed connection window. */
                BUG_ON(total > window_conn);

                if (!qfctl_rblocked(&qcc->tx.fc) &&
                    !qfctl_rblocked(&qcs->tx.fc) && window_conn > total) {
                        if ((ret = qcs_send(qcs, frms, window_conn - total)) < 0) {
                                /* Temporarily remove QCS from send-list. */
                                LIST_DEL_INIT(&qcs->el_send);
                                LIST_APPEND(qcs_failed, &qcs->el_send);
                                continue;
                        }

                        total += ret;
                        if (ret) {
                                /* Move QCS with some bytes transferred at the
                                 * end of send-list for next iterations.
                                 */
                                LIST_DEL_INIT(&qcs->el_send);
                                LIST_APPEND(&qcc->send_list, &qcs->el_send);
                                /* Remember first moved QCS as checkpoint to interrupt loop */
                                if (!first_qcs)
                                        first_qcs = qcs;
                        }
                }
        }

        TRACE_LEAVE(QMUX_EV_QCC_SEND, qcc->conn);
        return total;
}

/* Schedule <qcc> after emission was interrupted on pacing. */
static void qcc_wakeup_pacing(struct qcc *qcc)
{
        /* Sleep to be able to reemit at least a single packet */
        const int inter = qcc->tx.pacer.cc->algo->pacing_inter(qcc->tx.pacer.cc);
        /* Convert nano to milliseconds rounded up, with 1ms as minimal value. */
        const int expire = MAX((inter + 999999) / 1000000, 1);
        qcc->pacing_task->expire = tick_add_ifset(now_ms, MS_TO_TICKS(expire));
        ++qcc->tx.paced_sent_ctr;
}

/* Proceed to sending. Loop through all available streams for the <qcc>
 * instance and try to send as much as possible.
 *
 * Returns the total of bytes sent to the transport layer.
 */
static int qcc_io_send(struct qcc *qcc)
{
        struct list *frms = &qcc->tx.frms;
        /* Temporary list for QCS on error. */
        struct list qcs_failed = LIST_HEAD_INIT(qcs_failed);
        struct qcs *qcs, *qcs_tmp;
        uint64_t window_conn = qfctl_rcap(&qcc->tx.fc);
        int ret = 0, total = 0, resent;

        TRACE_ENTER(QMUX_EV_QCC_SEND, qcc->conn);

        if (qcc_is_pacing_active(qcc->conn)) {
                /* Always reset pacing_task timer to prevent unnecessary execution. */
                qcc->pacing_task->expire = TICK_ETERNITY;
        }

        /* TODO if socket in transient error, sending should be temporarily
         * disabled for all frames. However, checking for send subscription is
         * not valid as this may be caused by a congestion error which only
         * apply for STREAM frames.
         */

        /* Check for transport error. */
        if (qcc->flags & QC_CF_ERR_CONN || qcc->conn->flags & CO_FL_ERROR) {
                TRACE_DEVEL("connection on error", QMUX_EV_QCC_SEND, qcc->conn);
                goto out;
        }

        /* Check for locally detected connection error. */
        if (qcc->flags & QC_CF_ERRL) {
                /* Prepare a CONNECTION_CLOSE if not already done. */
                if (!(qcc->flags & QC_CF_ERRL_DONE)) {
                        TRACE_DATA("report a connection error", QMUX_EV_QCC_SEND|QMUX_EV_QCC_ERR, qcc->conn);
                        /* TODO implement a QMux alternative */
                        if (conn_is_quic(qcc->conn))
                                quic_set_connection_close(qcc->conn->handle.qc, qcc->err);
                        qcc->flags |= QC_CF_ERRL_DONE;
                }
                goto out;
        }

        if (qcc->conn->flags & CO_FL_SOCK_WR_SH) {
                qcc->conn->flags |= CO_FL_ERROR;
                TRACE_DEVEL("connection on error", QMUX_EV_QCC_SEND, qcc->conn);
                goto out;
        }

        if (qcc->app_st < QCC_APP_ST_INIT) {
                if (qcc_app_init(qcc))
                        goto out;
        }

        if (!LIST_ISEMPTY(&qcc->lfctl.frms)) {
                if (qcc_emit_fctl(qcc)) {
                        TRACE_DEVEL("flow-control frames rejected by transport, aborting send", QMUX_EV_QCC_SEND, qcc->conn);
                        goto out;
                }
        }

        if (qcc_emit_rs_ss(qcc)) {
                TRACE_DEVEL("emission interrupted on STOP_SENDING/RESET_STREAM send error", QMUX_EV_QCC_SEND, qcc->conn);
                goto out;
        }

        /* Encode new STREAM frames if list has been previously cleared. */
        if (LIST_ISEMPTY(frms) && !LIST_ISEMPTY(&qcc->send_list)) {
                total = qcc_build_frms(qcc, &qcs_failed);
                if (LIST_ISEMPTY(frms))
                        goto out;
        }

        if (!LIST_ISEMPTY(frms) && qcc_is_pacing_active(qcc->conn)) {
                if (!quic_pacing_reload(&qcc->tx.pacer)) {
                        qcc_wakeup_pacing(qcc);
                        total = 0;
                        goto out;
                }
        }

        /* Retry sending until no frame to send, data rejected or connection
         * flow-control limit reached.
         */
        while ((ret = qcc_send_frames(qcc, frms, 1)) == 0 && !qfctl_rblocked(&qcc->tx.fc)) {
                /* TODO should this check also be performed for QUIC ? */
                if (!conn_is_quic(qcc->conn) && (qcc->conn->flags & CO_FL_ERROR)) {
                        TRACE_DEVEL("connection on error", QMUX_EV_QCC_SEND, qcc->conn);
                        goto out;
                }

                window_conn = qfctl_rcap(&qcc->tx.fc);
                resent = 0;

                /* Reloop over <qcc.send_list>. Useful for streams which have
                 * fulfilled their qc_stream_desc buf and have now release it.
                 */
                list_for_each_entry_safe(qcs, qcs_tmp, &qcc->send_list, el_send) {
                        /* Only streams blocked on flow-control or waiting on a
                         * new qc_stream_desc should be present in send_list as
                         * long as transport layer can handle all data.
                         */
                        BUG_ON((!conn_is_quic(qcc->conn) || qcs->tx.stream->buf) &&
                               !qfctl_rblocked(&qcs->tx.fc));

                        /* Total sent bytes must not exceed connection window. */
                        BUG_ON(resent > window_conn);

                        if (!qfctl_rblocked(&qcs->tx.fc) && window_conn > resent) {
                                if ((ret = qcs_send(qcs, frms, window_conn - resent)) < 0) {
                                        LIST_DEL_INIT(&qcs->el_send);
                                        LIST_APPEND(&qcs_failed, &qcs->el_send);
                                        continue;
                                }

                                total += ret;
                                resent += ret;
                        }
                }
        }

        if (ret == 1) {
                /* qcc_send_frames cannot return 1 if pacing not used. */
                BUG_ON(!qcc_is_pacing_active(qcc->conn));
                qcc_wakeup_pacing(qcc);
        }

 out:
        /* Re-insert on-error QCS at the end of the send-list. */
        if (!LIST_ISEMPTY(&qcs_failed)) {
                list_for_each_entry_safe(qcs, qcs_tmp, &qcs_failed, el_send) {
                        LIST_DEL_INIT(&qcs->el_send);
                        LIST_APPEND(&qcc->send_list, &qcs->el_send);
                }

                if (!qfctl_rblocked(&qcc->tx.fc))
                        tasklet_wakeup(qcc->wait_event.tasklet);
        }

        if (qcc->conn->flags & CO_FL_ERROR && !(qcc->flags & QC_CF_ERR_CONN)) {
                TRACE_ERROR("error reported by transport layer",
                            QMUX_EV_QCC_SEND, qcc->conn);
                qcc->flags |= QC_CF_ERR_CONN;
        }

        TRACE_LEAVE(QMUX_EV_QCC_SEND, qcc->conn);
        return total;
}

/* Detects QUIC handshake completion. Any SE_FL_WAIT_FOR_HS streams are woken
 * up if wait-for-handshake is active.
 */
static void qcc_wait_for_hs(struct qcc *qcc)
{
        struct connection *conn = qcc->conn;
        struct quic_conn *qc = conn->handle.qc;
        struct eb64_node *node;
        struct qcs *qcs;

        TRACE_ENTER(QMUX_EV_QCC_RECV, qcc->conn);

        if (qc->state >= QUIC_HS_ST_COMPLETE) {
                if (conn->flags & CO_FL_EARLY_SSL_HS) {
                        TRACE_STATE("mark early data as ready", QMUX_EV_QCC_WAKE, conn);
                        conn->flags &= ~CO_FL_EARLY_SSL_HS;
                }

                /* wake-up any stream blocked on early data transfer */
                node = eb64_first(&qcc->streams_by_id);
                while (node) {
                        qcs = container_of(node, struct qcs, by_id);
                        if (qcs_sc(qcs) && se_fl_test(qcs->sd, SE_FL_WAIT_FOR_HS))
                                qcs_notify_recv(qcs);
                        node = eb64_next(node);
                }

                qcc->flags &= ~QC_CF_WAIT_HS;
        }

        TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
}

/* Proceed on receiving. Loop on streams subscribed in recv_list and performed
 * STREAM frames decoding upon them.
 *
 * Returns the number of newly transcoded bytes.
 */
static int qcc_io_recv(struct qcc *qcc)
{
        struct qcs *qcs;
        int total = 0, ret;

        TRACE_ENTER(QMUX_EV_QCC_RECV, qcc->conn);

        if (qcc->flags & QC_CF_ERRL) {
                TRACE_DATA("connection on error", QMUX_EV_QCC_RECV, qcc->conn);
                TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
                return 0;
        }

        if ((qcc->flags & QC_CF_WAIT_HS) && !(qcc->wait_event.events & SUB_RETRY_RECV))
                qcc_wait_for_hs(qcc);

        if (!conn_is_quic(qcc->conn)) {
                if (!(qcc->wait_event.events & SUB_RETRY_RECV))
                        qcc_qmux_recv(qcc);
        }

 next_recv:
        while (!LIST_ISEMPTY(&qcc->recv_list)) {
                qcs = LIST_ELEM(qcc->recv_list.n, struct qcs *, el_recv);
                /* No need to add an uni local stream in recv_list. */
                BUG_ON(quic_stream_is_uni(qcs->id) && quic_stream_is_local(qcc, qcs->id));

                while (qcs_rx_avail_data(qcs) && !(qcs->flags & QC_SF_DEM_FULL)) {
                        ret = qcc_decode_qcs(qcc, qcs);
                        if (ret <= 0) {
                                LIST_DEL_INIT(&qcs->el_recv);
                                /* Interrupt all receive if connection on error. */
                                if (qcc->flags & QC_CF_ERRL)
                                        goto done;
                                /* Decode next entry if stream on error. */
                                goto next_recv;
                        }

                        total += ret;
                }

                /* Always remove QCS from recv_list to prevent infinite loop.
                 * This is performed even if inner loop was not executed : QCS
                 * has nothing to do in recv_list if no avail Rx data or demux
                 * is blocked. Next decoding will be performed on new data read
                 * unless demux is blocked. In this case QCS will be reinserted
                 * in recv_list on unblocking to execute decode here again.
                 */
                LIST_DEL_INIT(&qcs->el_recv);
        }

 done:
        TRACE_LEAVE(QMUX_EV_QCC_RECV, qcc->conn);
        return total;
}

/* Calculate the number of bidirectional streams which can still be opened for
 * <conn> connection. This depends on flow-control set by the peer.
 *
 * Returns the value which is a positive integer or 0 if no new stream
 * currently available.
 */
static int qcm_avail_streams(struct connection *conn)
{
        const struct server *srv = __objt_server(conn->target);
        struct qcc *qcc = conn->ctx;
        int ret;

        if (!qcc_be_is_reusable(qcc))
                return 0;

        ret = qcc_fctl_avail_streams(qcc, 1);

        /* Enforce stream_max_concurrent limit even if peer allows more streams. */
        if (ret > quic_tune.be.stream_max_concurrent - qcc->nb_hreq)
                ret = quic_tune.be.stream_max_concurrent - qcc->nb_hreq;

        /* Now cap return value if reaching max-reuse server or maximum stream
         * ID. qcc_be_is_reusable() already detected if one of these has been
         * exceeded.
         */
        if (ret > 1 && srv->max_reuse >= 0 && qcc->tot_sc == srv->max_reuse)
                ret = 1;
        else if (ret > 1 && unlikely(qcc->next_bidi_l == QCS_ID_MAX_STRM_CL_BIDI))
                ret = 1;

        return ret;
}

/* Returns the number of streams currently attached into <conn> connection.
 * Used to determine if a connection can be considered as idle or not.
 */
static int qcm_used_streams(struct connection *conn)
{
        struct qcc *qcc = conn->ctx;
        return qcc->nb_sc;
}

/* Release all streams which have their transfer operation achieved. */
static void qcc_purge_streams(struct qcc *qcc)
{
        struct qcs *qcs;

        TRACE_ENTER(QMUX_EV_QCC_WAKE, qcc->conn);

        while (!LIST_ISEMPTY(&qcc->purg_list)) {
                qcs = LIST_ELEM(qcc->purg_list.n, struct qcs *, el_send);

                TRACE_STATE("purging stream", QMUX_EV_QCC_WAKE, qcs->qcc->conn, qcs);
                BUG_ON_HOT(!qcs_is_completed(qcs));
                qcs_destroy(qcs);
        }

        TRACE_LEAVE(QMUX_EV_QCC_WAKE, qcc->conn);
}

/* Loop through all qcs from <qcc> and wake their associated data layer if
 * still active. Also report error on it if connection is already in error.
 */
static void qcc_wake_streams(struct qcc *qcc)
{
        struct qcs *qcs;
        struct eb64_node *node;

        TRACE_POINT(QMUX_EV_QCC_WAKE, qcc->conn);

        for (node = eb64_first(&qcc->streams_by_id); node;
             node = eb64_next(node)) {
                qcs = eb64_entry(node, struct qcs, by_id);

                if (!qcs_sc(qcs))
                        continue;

                if (qcc->flags & (QC_CF_ERR_CONN|QC_CF_ERRL)) {
                        TRACE_POINT(QMUX_EV_QCC_WAKE, qcc->conn, qcs);
                        se_fl_set_error(qcs->sd);
                }

                qcs_alert(qcs);
        }
}

/* Conduct operations which should be made for <qcc> connection after
 * input/output. Most notably, closed streams are purged which may leave the
 * connection has ready to be released.
 *
 * Returns 1 if <qcc> must be released else 0.
 */
static int qcc_io_process(struct qcc *qcc)
{
        if (!LIST_ISEMPTY(&qcc->purg_list))
                qcc_purge_streams(qcc);

        /* Check if a soft-stop is in progress.
         *
         * TODO this is relevant for frontend connections only.
         */
        if (unlikely(qcc->proxy->flags & (PR_FL_DISABLED|PR_FL_STOPPED))) {
                int close = 1;

                /* If using listener socket, soft-stop is not supported. The
                 * connection must be closed immediately.
                 */
                if (conn_is_quic(qcc->conn) && !qc_test_fd(qcc->conn->handle.qc)) {
                        TRACE_DEVEL("proxy disabled with listener socket, closing connection", QMUX_EV_QCC_WAKE, qcc->conn);
                        qcc->conn->flags |= (CO_FL_SOCK_RD_SH|CO_FL_SOCK_WR_SH);
                        qcc_io_send(qcc);
                        goto out;
                }

                TRACE_DEVEL("proxy disabled, prepare connection soft-stop", QMUX_EV_QCC_WAKE, qcc->conn);

                /* If a close-spread-time option is set, we want to avoid
                 * closing all the active HTTP3 connections at once so we add a
                 * random factor that will spread the closing.
                 */
                if (tick_isset(global.close_spread_end)) {
                        int remaining_window = tick_remain(now_ms, global.close_spread_end);
                        if (remaining_window) {
                                /* This should increase the closing rate the
                                 * further along the window we are. */
                                close = (remaining_window <= statistical_prng_range(global.close_spread_time));
                        }
                }
                else if (global.tune.options & GTUNE_DISABLE_ACTIVE_CLOSE) {
                        close = 0; /* let the client close his connection himself */
                }

                if (close)
                        qcc_app_shutdown(qcc);
        }

        /* Report error if set on stream endpoint layer. */
        if (qcc->flags & (QC_CF_ERR_CONN|QC_CF_ERRL))
                qcc_wake_streams(qcc);

 out:
        if (qcc_is_dead(qcc))
                return 1;

        return 0;
}

/* Conduct I/O operations to finalize <qcc> app layer initialization. Note that
 * <qcc> app state may remain NULL even on success, if only a transient
 * blocking was encountered. Finalize operation can be retried later.
 *
 * Returns 0 on success else non-zero.
 */
static int qcc_app_init(struct qcc *qcc)
{
        int ret;

        TRACE_ENTER(QMUX_EV_QCC_SEND, qcc->conn);

        if (qcc->app_ops->finalize) {
                ret = qcc->app_ops->finalize(qcc->ctx);
                if (ret < 0) {
                        TRACE_ERROR("app ops finalize error", QMUX_EV_QCC_NEW, qcc->conn);
                        goto err;
                }

                if (ret) {
                        TRACE_STATE("cannot finalize app ops yet", QMUX_EV_QCC_NEW, qcc->conn);
                        goto again;
                }
        }

        qcc->app_st = QCC_APP_ST_INIT;

 again:
        TRACE_LEAVE(QMUX_EV_QCC_SEND, qcc->conn);
        return 0;

 err:
        TRACE_DEVEL("leaving on error", QMUX_EV_QCC_SEND, qcc->conn);
        return 1;
}

/* Execute application layer shutdown. If this operation is not defined, a
 * CONNECTION_CLOSE will be prepared as a fallback. This function is protected
 * against multiple invocations thanks to <qcc> application state context.
 */
static void qcc_app_shutdown(struct qcc *qcc)
{
        TRACE_ENTER(QMUX_EV_QCC_END, qcc->conn);

        if (qcc->flags & (QC_CF_ERR_CONN|QC_CF_ERRL)) {
                TRACE_DATA("connection on error", QMUX_EV_QCC_END, qcc->conn);
                goto out;
        }

        if (qcc->app_st >= QCC_APP_ST_SHUT)
                goto out;

        if (qcc->app_st < QCC_APP_ST_INIT) {
                if (qcc_app_init(qcc))
                        goto out;
        }

        TRACE_STATE("perform graceful shutdown", QMUX_EV_QCC_END, qcc->conn);
        qcc_report_term_evt(qcc, muxc_tevt_type_graceful_shut);
        if (qcc->app_ops && qcc->app_ops->shutdown) {
                qcc->app_ops->shutdown(qcc->ctx);
                qcc_io_send(qcc);
        }
        else {
                qcc->err = quic_err_transport(QC_ERR_NO_ERROR);
        }

        if (conn_is_quic(qcc->conn)) {
                /* Register "no error" code at transport layer. Do not use
                 * quic_set_connection_close() as retransmission may be performed to
                 * finalized transfers. Do not overwrite quic-conn existing code if
                 * already set.
                 *
                 * TODO implement a wrapper function for this in quic-conn module
                 */
                if (!(qcc->conn->handle.qc->flags & QUIC_FL_CONN_IMMEDIATE_CLOSE))
                        qcc->conn->handle.qc->err = qcc->err;
        }

        /* A connection is not reusable if app layer is closed. */
        if (qcc->flags & QC_CF_IS_BACK) {
                if (qcc->conn->flags & CO_FL_LIST_MASK)
                        conn_delete_from_tree(qcc->conn, tid);
                else if (qcc->conn->flags & CO_FL_SESS_IDLE)
                        session_unown_conn(qcc->conn->owner, qcc->conn);
        }

 out:
        qcc->app_st = QCC_APP_ST_SHUT;
        TRACE_LEAVE(QMUX_EV_QCC_END, qcc->conn);
}

/* Free all resources allocated for <qcc> connection. */
static void qcc_release(struct qcc *qcc)
{
        struct connection *conn = qcc->conn;
        struct eb64_node *node;
        struct quic_conn *qc;

        TRACE_ENTER(QMUX_EV_QCC_END, conn);

        task_destroy(qcc->pacing_task);

        if (qcc->task) {
                task_destroy(qcc->task);
                qcc->task = NULL;
        }

        /* free remaining qcs instances */
        node = eb64_first(&qcc->streams_by_id);
        while (node) {
                struct qcs *qcs = eb64_entry(node, struct qcs, by_id);
                node = eb64_next(node);
                qcs_free(qcs);
        }

        if (conn && conn_is_quic(conn)) {
                qc = conn->handle.qc;

                /* unsubscribe from all remaining qc_stream_desc */
                node = eb64_first(&qc->streams_by_id);
                while (node) {
                        struct qc_stream_desc *stream = eb64_entry(node, struct qc_stream_desc, by_id);
                        qc_stream_desc_sub_room(stream, NULL);
                        node = eb64_next(node);
                }

                /* register streams IDs so that quic-conn layer can ignore already closed streams. */
                if (!conn_is_back(conn)) {
                        qc->rx.stream_max_uni = qcc->largest_uni_r;
                        qc->rx.stream_max_bidi = qcc->largest_bidi_r;
                }
        }

        tasklet_free(qcc->wait_event.tasklet);
        if (conn) {
                qcc_report_term_evt(qcc, muxc_tevt_type_shutw);
                if (qcc->wait_event.events) {
                        conn->xprt->unsubscribe(conn, conn->xprt_ctx,
                                                qcc->wait_event.events,
                                                &qcc->wait_event);
                }
        }

        while (!LIST_ISEMPTY(&qcc->lfctl.frms)) {
                struct quic_frame *frm = LIST_ELEM(qcc->lfctl.frms.n, struct quic_frame *, list);
                qc_frm_free(conn_is_quic(qcc->conn) ? qcc->conn->handle.qc : NULL, &frm);
        }

        qcc_clear_frms(qcc);

        if (qcc->app_ops) {
                if (qcc->app_ops->release)
                        qcc->app_ops->release(qcc->ctx);
                if (conn && !conn_is_back(conn) && conn_is_quic(conn) && conn->handle.qc)
                        conn->handle.qc->strm_reject = qcc->app_ops->strm_reject;
        }
        TRACE_PROTO("application layer released", QMUX_EV_QCC_END, conn);

        if (!(qcc->flags & QC_CF_IS_BACK) && global.tune.streams_elasticity &&
            qcc->lfctl.ms_bidi_rel > 1) {
                _HA_ATOMIC_SUB(&tg_ctx->committed_extra_streams,
                               qcc->lfctl.ms_bidi_rel - 1);
        }

        if (conn && !conn_is_quic(conn)) {
                b_free(&qcc->rx.qmux_buf);
                b_free(&qcc->tx.qmux_buf);
                offer_buffers(NULL, 2);
        }

        TRACE_USER("release mux", QMUX_EV_QCC_END, conn);
        pool_free(pool_head_qcc, qcc);

        if (conn) {
                conn->mux = NULL;
                conn->ctx = NULL;

                TRACE_DEVEL("freeing conn", QMUX_EV_QCC_END, conn);

                conn_stop_tracking(conn);
                conn_full_close(conn);
                if (conn->destroy_cb)
                        conn->destroy_cb(conn);
                conn_free(conn);
        }

        TRACE_LEAVE(QMUX_EV_QCC_END);
}

struct task *qcc_io_cb(struct task *t, void *ctx, unsigned int state)
{
        struct qcc *qcc = ctx;
        struct connection *conn;
        int total = 0, conn_in_list;

        if (state & TASK_F_USR1) {
                /* the tasklet was idling on an idle connection, it might have
                 * been stolen, let's be careful!
                 */
                HA_SPIN_LOCK(IDLE_CONNS_LOCK, &idle_conns[tid].idle_conns_lock);
                if (t->context == NULL) {
                        /* The connection has been taken over by another thread,
                         * we're no longer responsible for it, so just free the
                         * tasklet, and do nothing.
                         */
                        HA_SPIN_UNLOCK(IDLE_CONNS_LOCK, &idle_conns[tid].idle_conns_lock);
                        tasklet_free((struct tasklet *)t);
                        t = NULL;
                        TRACE_LEAVE(QMUX_EV_QCC_WAKE);
                        return NULL;
                }
                conn = qcc->conn;
                TRACE_ENTER(QMUX_EV_QCC_WAKE, conn);

                /* Remove the connection from the list, to be sure nobody attempts
                 * to use it while we handle the I/O events
                 */
                if (LIST_INLIST(&conn->idle_list)) {
                        conn_in_list = 1;
                        BUG_ON(!(conn->flags & CO_FL_LIST_MASK));
                        conn_delete_from_tree(conn, tid);
                }
                else if (LIST_INLIST(&conn->sess_el)) {
                        conn_in_list = 1;
                        BUG_ON(!(conn->flags & CO_FL_SESS_IDLE));
                        /* session_detach_idle_conn cannot fail thanks to LIST_INLIST() above test */
                        session_detach_idle_conn(conn->owner, conn);
                }
                else {
                        conn_in_list = 0;
                }

                HA_SPIN_UNLOCK(IDLE_CONNS_LOCK, &idle_conns[tid].idle_conns_lock);
        } else {
                /* we're certain the connection was not in an idle list */
                conn = qcc->conn;
                TRACE_ENTER(QMUX_EV_QCC_WAKE, conn);
                conn_in_list = 0;
        }

        if (!(qcc->wait_event.events & SUB_RETRY_SEND))
                total += qcc_io_send(qcc);

        total += qcc_io_recv(qcc);

        if (qcc_io_process(qcc)) {
                TRACE_STATE("releasing dead connection", QMUX_EV_QCC_WAKE, conn);
                goto release;
        }

        if (total)
                qcc_refresh_timeout(qcc);

        /* Trigger pacing task is emission should be retried after some delay. */
        if (qcc_is_pacing_active(conn)) {
                if (tick_isset(qcc->pacing_task->expire))
                        task_queue(qcc->pacing_task);
        }

        if (conn_in_list) {
                struct server *srv = __objt_server(conn->target);

                if (srv->cur_admin & SRV_ADMF_MAINT) {
                        /* Do not store an idle conn if server in maintenance. */
                        goto release;
                }

                if (conn->flags & CO_FL_SESS_IDLE) {
                        if (!session_reinsert_idle_conn(conn->owner, conn)) {
                                /* session add conn failure */
                                goto release;
                        }
                }
                else {
                        srv_add_idle(srv, conn, (conn->flags & CO_FL_LIST_MASK) == CO_FL_SAFE_LIST);
                }

                /* Do not access conn without protection as soon as it is reinserted in idle list. */
                conn = NULL;
        }

        TRACE_LEAVE(QMUX_EV_QCC_WAKE, conn);

        return t;

 release:
        qcc_app_shutdown(qcc);
        qcc_release(qcc);

        TRACE_LEAVE(QMUX_EV_QCC_WAKE);

        return NULL;
}

static struct task *qcc_pacing_task(struct task *t, void *ctx, unsigned int state)
{
        struct qcc *qcc = ctx;
        int expired = tick_is_expired(t->expire, now_ms);

        TRACE_ENTER(QMUX_EV_QCC_WAKE, qcc->conn);

        if (!expired) {
                if (!tick_isset(t->expire))
                        TRACE_DEVEL("cancelled pacing task", QMUX_EV_QCC_WAKE, qcc->conn);
                goto requeue;
        }

        /* Reschedule I/O immediately. */
        tasklet_wakeup_after(NULL, qcc->wait_event.tasklet);

 requeue:
        TRACE_LEAVE(QMUX_EV_QCC_WAKE, qcc->conn);
        return t;
}

static struct task *qcc_timeout_task(struct task *t, void *ctx, unsigned int state)
{
        struct qcc *qcc = ctx;
        int expired = tick_is_expired(t->expire, now_ms);

        TRACE_ENTER(QMUX_EV_QCC_WAKE, qcc ? qcc->conn : NULL);

        if (qcc) {
                 /* Make sure nobody stole the connection from us */
                HA_SPIN_LOCK(IDLE_CONNS_LOCK, &idle_conns[tid].idle_conns_lock);

                /* Somebody already stole the connection from us, so we should
                 * not free it, we just have to free the task.
                 */
                if (!t->context) {
                        qcc = NULL;
                        HA_SPIN_UNLOCK(IDLE_CONNS_LOCK, &idle_conns[tid].idle_conns_lock);
                        goto out;
                }

                if (!expired) {
                        HA_SPIN_UNLOCK(IDLE_CONNS_LOCK, &idle_conns[tid].idle_conns_lock);
                        TRACE_DEVEL("not expired", QMUX_EV_QCC_WAKE, qcc->conn);
                        goto requeue;
                }

                if (!qcc_may_expire(qcc)) {
                        HA_SPIN_UNLOCK(IDLE_CONNS_LOCK, &idle_conns[tid].idle_conns_lock);
                        TRACE_DEVEL("cannot expired", QMUX_EV_QCC_WAKE, qcc->conn);
                        t->expire = TICK_ETERNITY;
                        goto requeue;
                }

                /* We're about to destroy the connection, so make sure nobody
                 * attempts to steal it from us.
                 */
                if (qcc->conn->flags & CO_FL_LIST_MASK)
                        conn_delete_from_tree(qcc->conn, tid);
                else if (qcc->conn->flags & CO_FL_SESS_IDLE)
                        session_unown_conn(qcc->conn->owner, qcc->conn);

                HA_SPIN_UNLOCK(IDLE_CONNS_LOCK, &idle_conns[tid].idle_conns_lock);
        }

        task_destroy(t);

        if (!qcc) {
                TRACE_DEVEL("no more qcc", QMUX_EV_QCC_WAKE);
                goto out;
        }

        /* Mark timeout as triggered by setting task to NULL. */
        qcc->task = NULL;

        /* TODO depending on the timeout condition, different shutdown mode
         * should be used. For http keep-alive or disabled proxy, a graceful
         * shutdown should occurs. For all other cases, an immediate close
         * seems legitimate.
         */
        TRACE_STATE("releasing dead connection", QMUX_EV_QCC_WAKE, qcc->conn);
        qcc_app_shutdown(qcc);
        qcc_release(qcc);

 out:
        TRACE_LEAVE(QMUX_EV_QCC_WAKE);
        return NULL;

 requeue:
        TRACE_LEAVE(QMUX_EV_QCC_WAKE);
        return t;
}

/* Minimal initialization of <qcc> members to use qcc_release() safely. */
static void _qcc_init(struct qcc *qcc)
{
        qcc->conn = NULL;
        qcc->pacing_task = NULL;
        qcc->task = NULL;
        qcc->wait_event.tasklet = NULL;
        qcc->app_ops = NULL;
        qcc->streams_by_id = EB_ROOT_UNIQUE;
        LIST_INIT(&qcc->lfctl.frms);
        LIST_INIT(&qcc->tx.frms);
}

static int qcm_init(struct connection *conn, struct proxy *prx,
                    struct session *sess, struct buffer *input)
{
        struct qcc *qcc;
        const struct quic_transport_params *lparams, *rparams;
        void *conn_ctx = conn->ctx;

        TRACE_ENTER(QMUX_EV_QCC_NEW);

        qcc = pool_alloc(pool_head_qcc);
        if (!qcc) {
                TRACE_ERROR("alloc failure", QMUX_EV_QCC_NEW);
                goto err;
        }

        _qcc_init(qcc);
        conn->ctx = qcc;
        qcc->nb_hreq = qcc->nb_sc = qcc->tot_sc = 0;
        qcc->flags = conn_is_back(conn) ? QC_CF_IS_BACK : 0;
        qcc->app_st = QCC_APP_ST_NULL;
        qcc->glitches = 0;
        qcc->term_evts_log = 0;
        qcc->err = quic_err_transport(QC_ERR_NO_ERROR);

        if (conn_is_quic(conn)) {
                /* Server parameters, params used for RX flow control. */
                lparams = &conn->handle.qc->rx.params;

                qcc->lfctl.ms_bidi = qcc->lfctl.ms_bidi_init = qcc->lfctl.ms_bidi_rel = lparams->initial_max_streams_bidi;
                qcc->lfctl.ms_uni = lparams->initial_max_streams_uni;
                qcc->lfctl.msd_bidi_l = lparams->initial_max_stream_data_bidi_local;
                qcc->lfctl.msd_bidi_r = lparams->initial_max_stream_data_bidi_remote;
                qcc->lfctl.msd_uni_r = lparams->initial_max_stream_data_uni;
                qcc->lfctl.cl_bidi_r = 0;

                qcc->lfctl.md = qcc->lfctl.md_init = lparams->initial_max_data;
                qcc->lfctl.offsets_recv = qcc->lfctl.offsets_consume = 0;

                rparams = &conn->handle.qc->tx.params;
                qfctl_init(&qcc->tx.fc, rparams->initial_max_data);
                qcc->rfctl.ms_uni = rparams->initial_max_streams_uni;
                qcc->rfctl.ms_bidi = rparams->initial_max_streams_bidi;
                qcc->rfctl.msd_bidi_l = rparams->initial_max_stream_data_bidi_local;
                qcc->rfctl.msd_bidi_r = rparams->initial_max_stream_data_bidi_remote;
                qcc->rfctl.msd_uni_l = rparams->initial_max_stream_data_uni;
        }
        else {
                rparams = xprt_qmux_rparams(conn->xprt_ctx);
                qfctl_init(&qcc->tx.fc, rparams->initial_max_data);

                qcc->rfctl.ms_uni = rparams->initial_max_streams_uni;
                qcc->rfctl.ms_bidi = rparams->initial_max_streams_bidi;
                qcc->rfctl.msd_bidi_l = rparams->initial_max_stream_data_bidi_local;
                qcc->rfctl.msd_bidi_r = rparams->initial_max_stream_data_bidi_remote;
                qcc->rfctl.msd_uni_l = rparams->initial_max_stream_data_uni;

                lparams = xprt_qmux_lparams(conn->xprt_ctx);
                qcc->lfctl.ms_bidi = qcc->lfctl.ms_bidi_init = qcc->lfctl.ms_bidi_rel= lparams->initial_max_streams_bidi;
                qcc->lfctl.ms_uni = lparams->initial_max_streams_uni;
                qcc->lfctl.msd_bidi_l = lparams->initial_max_stream_data_bidi_local;
                qcc->lfctl.msd_bidi_r = lparams->initial_max_stream_data_bidi_remote;
                qcc->lfctl.msd_uni_r = lparams->initial_max_stream_data_uni;
                qcc->lfctl.cl_bidi_r = 0;

                qcc->lfctl.md = qcc->lfctl.md_init = lparams->initial_max_data;
                qcc->lfctl.offsets_recv = qcc->lfctl.offsets_consume = 0;
        }

        qcc->tx.buf_in_flight = 0;

        if (qcc_is_pacing_active(conn)) {
                quic_pacing_init(&qcc->tx.pacer, &conn->handle.qc->path->cc);
                qcc->tx.paced_sent_ctr = 0;

                /* Initialize pacing_task. */
                qcc->pacing_task = task_new_here();
                if (!qcc->pacing_task) {
                        TRACE_ERROR("pacing task alloc failure", QMUX_EV_QCC_NEW);
                        goto err;
                }
                qcc->pacing_task->process = qcc_pacing_task;
                qcc->pacing_task->context = qcc;
                qcc->pacing_task->expire = TICK_ETERNITY;
                qcc->pacing_task->state |= TASK_F_WANTS_TIME;
        }

        if (!conn_is_quic(conn)) {
                qcc->tx.qmux_buf = BUF_NULL;
                qcc->rx.qmux_buf = BUF_NULL;

                /* Rx buffer is transferred from xprt layer - necessary if too many data were read */
                qcc->rx.rlen = xprt_qmux_xfer_rxbuf(conn->xprt_ctx, &qcc->rx.qmux_buf);
                /* Cannot have a non empty record with an empty buffer. */
                BUG_ON(qcc->rx.rlen && !b_data(&qcc->rx.qmux_buf));
        }

        if (conn_is_back(conn)) {
                qcc->next_bidi_l    = 0x00;
                qcc->largest_bidi_r = 0x01;
                qcc->next_uni_l     = 0x02;
                qcc->largest_uni_r  = 0x03;
        }
        else {
                qcc->largest_bidi_r = 0x00;
                qcc->next_bidi_l    = 0x01;
                qcc->largest_uni_r  = 0x02;
                qcc->next_uni_l     = 0x03;

                if (global.tune.streams_elasticity && qcc->lfctl.ms_bidi_init > 1) {
                        _HA_ATOMIC_ADD(&tg_ctx->committed_extra_streams,
                                       qcc->lfctl.ms_bidi_init - 1);
                }
        }

        qcc->wait_event.tasklet = tasklet_new();
        if (!qcc->wait_event.tasklet) {
                TRACE_ERROR("tasklet alloc failure", QMUX_EV_QCC_NEW);
                goto err;
        }

        LIST_INIT(&qcc->recv_list);
        LIST_INIT(&qcc->send_list);
        LIST_INIT(&qcc->fctl_list);
        LIST_INIT(&qcc->buf_wait_list);
        LIST_INIT(&qcc->purg_list);

        qcc->wait_event.tasklet->process = qcc_io_cb;
        qcc->wait_event.tasklet->context = qcc;
        qcc->wait_event.tasklet->state  |= TASK_F_WANTS_TIME;
        qcc->wait_event.events = 0;

        qcc->proxy = prx;
        /* haproxy timeouts */
        if (conn_is_back(conn)) {
                qcc->timeout = prx->timeout.server;
                qcc->shut_timeout = tick_isset(prx->timeout.serverfin) ?
                                    prx->timeout.serverfin : prx->timeout.server;
        }
        else {
                qcc->timeout = prx->timeout.client;
                qcc->shut_timeout = tick_isset(prx->timeout.clientfin) ?
                                    prx->timeout.clientfin : prx->timeout.client;
        }

        /* Always allocate task even if timeout is unset. In MUX code, if task
         * is NULL, it indicates that a timeout has stroke earlier.
         */
        qcc->task = task_new_here();
        if (!qcc->task) {
                TRACE_ERROR("timeout task alloc failure", QMUX_EV_QCC_NEW);
                goto err;
        }
        qcc->task->process = qcc_timeout_task;
        qcc->task->context = qcc;
        qcc->task->expire = tick_add_ifset(now_ms, qcc->timeout);

        qcc_reset_idle_start(qcc);
        LIST_INIT(&qcc->opening_list);

        if (conn_is_quic(conn)) {
                HA_ATOMIC_STORE(&conn->handle.qc->qcc, qcc);
        }
        else {
                /* Wakeup MUX immediately if data copied from XPRT layer. */
                if (unlikely(b_data(&qcc->rx.qmux_buf)))
                        tasklet_wakeup(qcc->wait_event.tasklet);
        }

        /* Register conn as app_ops may use it. */
        qcc->conn = conn;

        if (qcc_install_app_ops(qcc)) {
                TRACE_PROTO("Cannot install app layer", QMUX_EV_QCC_NEW|QMUX_EV_QCC_ERR, conn);
                goto err;
        }

        if (qcc->app_ops == &h3_ops && !conn_is_back(conn))
                proxy_inc_fe_cum_sess_ver_ctr(sess->listener, prx, 3);

        if (!conn_is_back(conn)) {
                /* Register conn for idle front closing. */
                LIST_APPEND(&mux_stopping_data[tid].list, &conn->stopping_list);

                if (tick_isset(qcc->task->expire)) {
                        /* Activate client timeout until the first bytes of data are received. */
                        TRACE_DEVEL("activate default timeout", QMUX_EV_QCC_NEW, conn);
                        task_queue(qcc->task);
                }

                /* init read cycle */
                tasklet_wakeup(qcc->wait_event.tasklet);

                if (conn_is_quic(conn)) {
                        /* MUX is initialized before QUIC handshake completion if early data
                         * received. Flag connection to delay stream processing if
                         * wait-for-handshake is active.
                         */
                        if (conn->handle.qc->state < QUIC_HS_ST_COMPLETE) {
                                if (!(conn->flags & CO_FL_EARLY_SSL_HS)) {
                                        TRACE_STATE("flag connection with early data", QMUX_EV_QCC_WAKE, conn);
                                        conn->flags |= CO_FL_EARLY_SSL_HS;
                                        /* subscribe for handshake completion */
                                        conn->xprt->subscribe(conn, conn->xprt_ctx, SUB_RETRY_RECV,
                                                              &qcc->wait_event);
                                        qcc->flags |= QC_CF_WAIT_HS;
                                }
                        }
                }
        }
        else {
                /* Initiate backend side transfer by creating the first
                 * bidirectional stream. MUX will then be woken up on QUIC
                 * handshake completion so that stream layer can start the
                 * transfer itself.
                 */
                struct qcs *qcs;
                struct stconn *sc = conn_ctx;

                /* TODO how to properly handle initial value of max-bidi-stream set to 0 ? */
                if (!qcc_fctl_avail_streams(qcc, 1)) {
                        TRACE_ERROR("peer does not allow any bidi stream, not yet supported",
                                    QMUX_EV_QCC_NEW|QMUX_EV_QCC_ERR, qcc->conn);
                        goto err;
                }

                qcs = qcc_init_stream_local(qcc, 1);
                if (!qcs) {
                        TRACE_PROTO("Cannot allocate a new locally initiated stream",
                                    QMUX_EV_QCC_NEW|QMUX_EV_QCC_ERR, conn);
                        goto err;
                }

                if (sc_attach_mux(sc, qcs, conn)) {
                        TRACE_ERROR("Cannot attach backend first init stream",
                                    QMUX_EV_QCC_NEW|QMUX_EV_QCC_ERR, conn);
                        qcs_free(qcs);
                        goto err;
                }

                qcs->sess = sess;
                qcs->sd = sc->sedesc;
                qcc->nb_sc++;
                qcc->tot_sc++;
        }

        TRACE_USER("init mux", QMUX_EV_QCC_NEW, conn);
        TRACE_LEAVE(QMUX_EV_QCC_NEW, conn);
        return 0;

 err:
        if (conn_is_quic(conn)) {
                /* Prepare CONNECTION_CLOSE, using INTERNAL_ERROR as fallback code if unset. */
                if (!(conn->handle.qc->flags & QUIC_FL_CONN_IMMEDIATE_CLOSE)) {
                        struct quic_err err = qcc && qcc->err.code ?
                          qcc->err : quic_err_transport(QC_ERR_INTERNAL_ERROR);
                        quic_set_connection_close(conn->handle.qc, err);
                }
        }

        if (qcc) {
                /* In case of MUX init failure, session will ensure connection is freed. */
                qcc->conn = NULL;
                qcc_release(qcc);
                conn->ctx = NULL;
        }

        TRACE_DEVEL("leaving on error", QMUX_EV_QCC_NEW, qcc ? conn : NULL);
        return -1;
}

static void qcm_destroy(void *ctx)
{
        struct qcc *qcc = ctx;

        TRACE_ENTER(QMUX_EV_QCC_END, qcc->conn);
        qcc_release(qcc);
        TRACE_LEAVE(QMUX_EV_QCC_END);
}

static int qcm_strm_attach(struct connection *conn, struct sedesc *sd, struct session *sess)
{
        struct qcs *qcs;
        struct qcc *qcc = conn->ctx;

        TRACE_ENTER(QMUX_EV_QCS_NEW, conn);

        /* Flow control limit on bidi streams should already have
         * been checked by a prior qcm_avail_streams() invocation.
         */
        BUG_ON(!qcc_fctl_avail_streams(qcc, 1));

        /* Connection should not be reused if already on error/closed. */
        BUG_ON(qcc->flags & QC_CF_ERRL || qcc->app_st >= QCC_APP_ST_SHUT);

        qcs = qcc_init_stream_local(qcc, 1);
        if (!qcs) {
                TRACE_DEVEL("leaving on error", QMUX_EV_QCS_NEW, qcc->conn);
                return -1;
        }

        if (sc_attach_mux(sd->sc, qcs, conn)) {
                TRACE_DEVEL("leaving on error", QMUX_EV_QCS_NEW, qcc->conn);
                qcs_free(qcs);
                return -1;
        }

        qcs->sess = sess;
        qcs->sd = sd->sc->sedesc;
        qcc->nb_sc++;
        qcc->tot_sc++;

        /* the connection is not idle anymore, let's mark this */
        HA_ATOMIC_AND(&qcc->wait_event.tasklet->state, ~TASK_F_USR1);
        xprt_set_used(qcc->conn, qcc->conn->xprt, qcc->conn->xprt_ctx);

        TRACE_LEAVE(QMUX_EV_QCS_NEW, conn);
        return 0;
}

static void qcm_strm_detach(struct sedesc *sd)
{
        struct qcs *qcs = sd->se;
        struct qcc *qcc = qcs->qcc;
        struct connection *conn = qcc->conn;
        struct session *sess = qcs->sess;

        TRACE_ENTER(QMUX_EV_STRM_END, conn, qcs);

        /* TODO this BUG_ON_HOT() is not correct as the stconn layer may detach
         * from the stream even if it is not closed remotely at the QUIC layer.
         * This happens for example when a stream must be closed due to a
         * rejected request. To better handle these cases, it will be required
         * to implement shutr/shutw MUX operations. Once this is done, this
         * BUG_ON_HOT() statement can be adjusted.
         */
        //BUG_ON_HOT(!qcs_is_close_remote(qcs));

        qcc_rm_sc(qcc);

        /* Prevent QCS free if there is remaining data to send. */
        if ((!qcs_is_close_local(qcs) || (qcs->flags & QC_SF_TO_STOP_SENDING)) &&
            !(qcc->flags & (QC_CF_ERR_CONN|QC_CF_ERRL))) {
                TRACE_STATE("remaining data, detaching qcs", QMUX_EV_STRM_END, conn, qcs);
                qcs->flags |= QC_SF_DETACH;
                qcc_refresh_timeout(qcc);

                /* TODO should detached QCS be ignored to consider conn as idle ? */
                COUNT_IF_HOT(conn_is_back(conn),
                             "QCS in detached state on BE side, may lower reuse rate");

                TRACE_LEAVE(QMUX_EV_STRM_END, qcc->conn, qcs);
                return;
        }

        qcs_destroy(qcs);

        /* Backend connection can be reused unless it is already on error/closed. */
        if ((qcc->flags & QC_CF_IS_BACK) && !qcc_is_dead(qcc) &&
            qcc->app_st == QCC_APP_ST_INIT) {
                if (conn->flags & CO_FL_PRIVATE) {
                        TRACE_DEVEL("handle private connection reuse", QMUX_EV_STRM_END, conn);

                        /* Ensure conn is attached into session. Most of the times
                         * this is already done during connect so this is a no-op.
                         */
                        session_add_conn(sess, conn);

                        if (!qcc->nb_sc) {
                                if (!LIST_INLIST(&conn->sess_el)) {
                                        /* Session insertion above has failed and connection is idle, remove it. */
                                        goto release;
                                }

                                /* mark that the tasklet may lose its context to another thread and
                                 * that the handler needs to check it under the idle conns lock.
                                 */
                                HA_ATOMIC_OR(&qcc->wait_event.tasklet->state, TASK_F_USR1);
                                xprt_set_idle(qcc->conn, qcc->conn->xprt, qcc->conn->xprt_ctx);

                                /* Ensure session can keep a new idle connection. */
                                if (session_check_idle_conn(sess, conn)) {
                                        TRACE_DEVEL("idle conn rejected by session", QMUX_EV_STRM_END, conn);
                                        goto release;
                                }
                                /* At this point, the connection is inserted into
                                 * session list and marked as idle, so it may already
                                 * have been purged from another thread.
                                 */
                                conn = NULL;
                                goto end;
                        }
                }
                else {
                        if (!qcc->nb_sc) {
                                TRACE_DEVEL("prepare for idle connection reuse", QMUX_EV_STRM_END, conn);

                                /* mark that the tasklet may lose its context to another thread and
                                 * that the handler needs to check it under the idle conns lock.
                                 */
                                HA_ATOMIC_OR(&qcc->wait_event.tasklet->state, TASK_F_USR1);
                                xprt_set_idle(qcc->conn, qcc->conn->xprt, qcc->conn->xprt_ctx);

                                if (!srv_add_to_idle_list(objt_server(conn->target), conn, 1)) {
                                        /* Idle conn insert failure, gracefully close the connection. */
                                        TRACE_DEVEL("idle connection cannot be kept on the server", QMUX_EV_STRM_END, conn);
                                        goto release;
                                }

                                /* At this point, the connection has been added to the
                                 * server idle list, so another thread may already have
                                 * hijacked it, so we can't do anything with it.
                                 */
                                conn = NULL;
                                goto end;
                        }
                        else if (!ceb_intree(&conn->hash_node.node) &&
                                 qcm_avail_streams(conn) &&
                                 objt_server(conn->target)) {
                                TRACE_DEVEL("mark connection as available for reuse", QMUX_EV_STRM_END, conn);
                                srv_add_to_avail_list(__objt_server(conn->target), conn);
                        }
                }
        }

        if (qcc_is_dead(qcc)) {
                TRACE_STATE("killing dead connection", QMUX_EV_STRM_END, conn);
                goto release;
        }
        else {
                TRACE_DEVEL("refreshing connection's timeout", QMUX_EV_STRM_END, conn);
                qcc_refresh_timeout(qcc);
        }

 end:
        TRACE_LEAVE(QMUX_EV_STRM_END, conn);
        return;

 release:
        qcc_app_shutdown(qcc);
        qcc_release(qcc);
        TRACE_LEAVE(QMUX_EV_STRM_END);
        return;
}

/* Called from the upper layer, to receive data */
static size_t qcm_strm_rcv_buf(struct stconn *sc, struct buffer *buf,
                               size_t count, int flags)
{
        struct qcs *qcs = __sc_mux_strm(sc);
        struct qcc *qcc = qcs->qcc;
        size_t ret = 0;
        char fin = 0;

        TRACE_ENTER(QMUX_EV_STRM_RECV, qcc->conn, qcs);

        ret = qcs_http_rcv_buf(qcs, buf, count, &fin);

        if (b_data(&qcs->rx.app_buf)) {
                se_fl_set(qcs->sd, SE_FL_RCV_MORE | SE_FL_WANT_ROOM);
        }
        else {
                se_fl_clr(qcs->sd, SE_FL_RCV_MORE | SE_FL_WANT_ROOM);

                /* Set end-of-input when full message properly received. */
                if (fin) {
                        TRACE_STATE("report end-of-input", QMUX_EV_STRM_RECV, qcc->conn, qcs);
                        se_fl_set(qcs->sd, SE_FL_EOI);

                        if (!(qcc->flags & QC_CF_IS_BACK)) {
                                /* If request EOM is reported to the upper layer, it means the
                                 * QCS now expects data from the opposite side.
                                 */
                                se_expect_data(qcs->sd);
                        }
                }

                /* Set end-of-stream on read closed. */
                if (qcs->flags & QC_SF_RECV_RESET ||
                    qcc->conn->flags & CO_FL_SOCK_RD_SH) {
                        TRACE_STATE("report end-of-stream", QMUX_EV_STRM_RECV, qcc->conn, qcs);
                        se_fl_set(qcs->sd, SE_FL_EOS);

                        /* Set error if EOI not reached. This may happen on
                         * RESET_STREAM reception or connection error.
                         */
                        if (!se_fl_test(qcs->sd, SE_FL_EOI)) {
                                TRACE_STATE("report error on stream aborted", QMUX_EV_STRM_RECV, qcc->conn, qcs);
                                se_fl_set(qcs->sd, SE_FL_ERROR);
                                se_report_term_evt(qcs->sd, (qcc->flags & QC_CF_ERR_CONN ? se_tevt_type_truncated_rcv_err : se_tevt_type_truncated_eos));
                        }
                        else {
                                se_report_term_evt(qcs->sd, (qcc->flags & QC_CF_ERR_CONN ? se_tevt_type_rcv_err : se_tevt_type_eos));
                        }
                }

                if (se_fl_test(qcs->sd, SE_FL_ERR_PENDING)) {
                        TRACE_STATE("report error", QMUX_EV_STRM_RECV, qcc->conn, qcs);
                        se_fl_set(qcs->sd, SE_FL_ERROR);
                }

                if (b_size(&qcs->rx.app_buf)) {
                        b_free(&qcs->rx.app_buf);
                        offer_buffers(NULL, 1);
                }
        }

        /* Restart demux if it was interrupted on full buffer. */
        if (ret && qcs->flags & QC_SF_DEM_FULL) {
                /* Ensure DEM_FULL is only set if there is available data to
                 * ensure we never do unnecessary wakeup here.
                 */
                BUG_ON(!qcs_rx_avail_data(qcs));

                qcs->flags &= ~QC_SF_DEM_FULL;
                if (!(qcc->flags & QC_CF_ERRL)) {
                        LIST_APPEND(&qcc->recv_list, &qcs->el_recv);
                        tasklet_wakeup(qcc->wait_event.tasklet);
                }
        }

        TRACE_LEAVE(QMUX_EV_STRM_RECV, qcc->conn, qcs);

        return ret;
}

static size_t qcm_strm_snd_buf(struct stconn *sc, struct buffer *buf,
                               size_t count, int flags)
{
        struct qcs *qcs = __sc_mux_strm(sc);
        const size_t old_data = qcs_prep_bytes(qcs);
        size_t ret = 0;
        char fin;

        TRACE_ENTER(QMUX_EV_STRM_SEND, qcs->qcc->conn, qcs);

        /* stream layer has been detached so no transfer must occur after. */
        BUG_ON_HOT(qcs->flags & QC_SF_DETACH);

        /* Report error if set on stream endpoint layer. */
        if (qcs->qcc->flags & (QC_CF_ERR_CONN|QC_CF_ERRL)) {
                se_fl_set(qcs->sd, SE_FL_ERROR);
                TRACE_DEVEL("connection in error", QMUX_EV_STRM_SEND, qcs->qcc->conn, qcs);
                goto end;
        }

        /* Cannot emit data after FIN/RESET_STREAM, drain extra payload. */
        if (qcs_is_close_local(qcs) || (qcs->flags & QC_SF_TO_RESET)) {
                BUG_ON_STRESS(1);
                ret = qcs_http_reset_buf(qcs, buf, count);
                goto end;
        }

        if (LIST_INLIST(&qcs->el_buf)) {
                TRACE_DEVEL("leaving on no buf avail", QMUX_EV_STRM_SEND, qcs->qcc->conn, qcs);
                goto end;
        }

        if (qfctl_sblocked(&qcs->qcc->tx.fc)) {
                TRACE_DEVEL("leaving on connection flow control",
                            QMUX_EV_STRM_SEND, qcs->qcc->conn, qcs);
                if (!LIST_INLIST(&qcs->el_fctl)) {
                        TRACE_DEVEL("append to fctl-list",
                                    QMUX_EV_STRM_SEND, qcs->qcc->conn, qcs);
                        LIST_APPEND(&qcs->qcc->fctl_list, &qcs->el_fctl);
                        tot_time_start(&qcs->timer.fctl);
                }
                goto end;
        }

        if (qfctl_sblocked(&qcs->tx.fc)) {
                TRACE_DEVEL("leaving on flow-control reached",
                            QMUX_EV_STRM_SEND, qcs->qcc->conn, qcs);
                tot_time_start(&qcs->timer.fctl);
                goto end;
        }

        ret = qcs_http_snd_buf(qcs, buf, count, &fin);
        if (fin) {
                TRACE_STATE("reached stream fin", QMUX_EV_STRM_SEND, qcs->qcc->conn, qcs);
                qcs->flags |= QC_SF_FIN_STREAM;
        }

        if (ret || fin) {
                const size_t data = qcs_prep_bytes(qcs) - old_data;
                if (data || fin)
                        qcc_send_stream(qcs, 0, data);

                /* Wake up MUX to emit newly transferred data. */
                if (!(qcs->qcc->wait_event.events & SUB_RETRY_SEND))
                        tasklet_wakeup(qcs->qcc->wait_event.tasklet);
        }

 end:
        TRACE_LEAVE(QMUX_EV_STRM_SEND, qcs->qcc->conn, qcs);

        return ret;
}


static size_t qcm_strm_nego_ff(struct stconn *sc, struct buffer *input,
                               size_t count, unsigned int flags)
{
        struct qcs *qcs = __sc_mux_strm(sc);
        size_t ret = 0;

        TRACE_ENTER(QMUX_EV_STRM_SEND, qcs->qcc->conn, qcs);

        /* stream layer has been detached so no transfer must occur after. */
        BUG_ON_HOT(qcs->flags & QC_SF_DETACH);

        if (!qcs->qcc->app_ops->nego_ff || !qcs->qcc->app_ops->done_ff) {
                /* Fast forwarding is not supported by the QUIC application layer */
                qcs->sd->iobuf.flags |= IOBUF_FL_NO_FF;
                goto end;
        }

        if (qcs->qcc->flags & (QC_CF_ERR_CONN|QC_CF_ERRL)) {
                /* Disable fast-forward if connection is on error. Eventually,
                 * error will be reported to stream-conn if snd_buf is invoked.
                 */
                TRACE_DEVEL("connection in error", QMUX_EV_STRM_SEND, qcs->qcc->conn, qcs);
                qcs->sd->iobuf.flags |= IOBUF_FL_NO_FF;
                goto end;
        }

        if (qcs_is_close_local(qcs) || (qcs->flags & QC_SF_TO_RESET)) {
                /* Cannot emit any new data if stream already closed. Data
                 * draining will be performed via snd_buf.
                 */
                TRACE_DEVEL("stream already closed", QMUX_EV_STRM_SEND, qcs->qcc->conn, qcs);
                qcs->sd->iobuf.flags |= IOBUF_FL_NO_FF;
                goto end;
        }

        if (LIST_INLIST(&qcs->el_buf)) {
                TRACE_DEVEL("leaving on no buf avail", QMUX_EV_STRM_SEND, qcs->qcc->conn, qcs);
                qcs->sd->iobuf.flags |= IOBUF_FL_FF_BLOCKED;
                goto end;
        }

        if (qfctl_sblocked(&qcs->qcc->tx.fc)) {
                TRACE_DEVEL("leaving on connection flow control", QMUX_EV_STRM_SEND, qcs->qcc->conn, qcs);
                if (!LIST_INLIST(&qcs->el_fctl)) {
                        TRACE_DEVEL("append to fctl-list", QMUX_EV_STRM_SEND, qcs->qcc->conn, qcs);
                        LIST_APPEND(&qcs->qcc->fctl_list, &qcs->el_fctl);
                }
                qcs->sd->iobuf.flags |= IOBUF_FL_FF_BLOCKED;
                goto end;
        }

        if (qfctl_sblocked(&qcs->tx.fc)) {
                TRACE_DEVEL("leaving on flow-control reached", QMUX_EV_STRM_SEND, qcs->qcc->conn, qcs);
                qcs->sd->iobuf.flags |= IOBUF_FL_FF_BLOCKED;
                goto end;
        }

        /* Alawys disable splicing */
        qcs->sd->iobuf.flags |= IOBUF_FL_NO_SPLICING;

        ret = qcs->qcc->app_ops->nego_ff(qcs, count);
        if (!ret)
                goto end;

        /* forward remaining input data */
        if (b_data(input)) {
                size_t xfer = ret;

                if (xfer > b_data(input))
                        xfer = b_data(input);
                b_add(qcs->sd->iobuf.buf, qcs->sd->iobuf.offset);
                qcs->sd->iobuf.data = b_xfer(qcs->sd->iobuf.buf, input, xfer);
                b_sub(qcs->sd->iobuf.buf, qcs->sd->iobuf.offset);

                /* Cannot forward more data, wait for room */
                if (b_data(input)) {
                        ret = 0;
                        goto end;
                }
        }
        ret -= qcs->sd->iobuf.data;

 end:
        TRACE_LEAVE(QMUX_EV_STRM_SEND, qcs->qcc->conn, qcs);
        return ret;
}

static size_t qcm_strm_done_ff(struct stconn *sc)
{
        struct qcs *qcs = __sc_mux_strm(sc);
        struct qcc *qcc = qcs->qcc;
        struct sedesc *sd = qcs->sd;
        size_t total = 0, data = sd->iobuf.data;

        TRACE_ENTER(QMUX_EV_STRM_SEND, qcs->qcc->conn, qcs);

        if (sd->iobuf.flags & IOBUF_FL_EOI) {
                TRACE_STATE("reached stream fin", QMUX_EV_STRM_SEND, qcs->qcc->conn, qcs);
                qcs->flags |= QC_SF_FIN_STREAM;
        }

        if (!(qcs->flags & QC_SF_FIN_STREAM) && !sd->iobuf.data) {
                TRACE_STATE("no data sent", QMUX_EV_STRM_SEND, qcs->qcc->conn, qcs);

                /* There is nothing to forward and the SD was blocked after a
                 * successful nego by the producer. We can try to release the
                 * TXBUF to retry. In this case, the TX buf MUST exist.
                 */
                if ((qcs->sd->iobuf.flags & IOBUF_FL_FF_WANT_ROOM) && !qcc_release_stream_txbuf(qcs))
                        qcs->sd->iobuf.flags &= ~(IOBUF_FL_FF_BLOCKED|IOBUF_FL_FF_WANT_ROOM);
                goto end;
        }

        total = qcs->qcc->app_ops->done_ff(qcs);
        if (total || qcs->flags & QC_SF_FIN_STREAM)
                qcc_send_stream(qcs, 0, total);

        /* Reset stconn iobuf information. */
        qcs->sd->iobuf.buf = NULL;
        qcs->sd->iobuf.offset = 0;
        qcs->sd->iobuf.data = 0;

        if (!(qcs->qcc->wait_event.events & SUB_RETRY_SEND))
                tasklet_wakeup(qcc->wait_event.tasklet);

  end:
        TRACE_LEAVE(QMUX_EV_STRM_SEND, qcs->qcc->conn, qcs);
        return data;
}

static int qcm_strm_resume_ff(struct stconn *sc, unsigned int flags)
{
        return 0;
}

/* Called from the upper layer, to subscribe <es> to events <event_type>. The
 * event subscriber <es> is not allowed to change from a previous call as long
 * as at least one event is still subscribed. The <event_type> must only be a
 * combination of SUB_RETRY_RECV and SUB_RETRY_SEND. It always returns 0.
 */
static int qcm_strm_subscribe(struct stconn *sc, int event_type,
                              struct wait_event *es)
{
        return qcs_subscribe(__sc_mux_strm(sc), event_type, es);
}

/* Called from the upper layer, to unsubscribe <es> from events <event_type>.
 * The <es> pointer is not allowed to differ from the one passed to the
 * subscribe() call. It always returns zero.
 */
static int qcm_strm_unsubscribe(struct stconn *sc, int event_type, struct wait_event *es)
{
        struct qcs *qcs = __sc_mux_strm(sc);

        BUG_ON(event_type & ~(SUB_RETRY_SEND|SUB_RETRY_RECV));
        BUG_ON(qcs->subs && qcs->subs != es);

        es->events &= ~event_type;
        if (!es->events)
                qcs->subs = NULL;

        return 0;
}

static int qcm_wake(struct connection *conn)
{
        struct qcc *qcc = conn->ctx;

        TRACE_ENTER(QMUX_EV_QCC_WAKE, conn);

        if (qcc_io_process(qcc)) {
                TRACE_STATE("releasing dead connection", QMUX_EV_QCC_WAKE, qcc->conn);
                goto release;
        }

        /* Wake all streams, unless an error is set as qcc_io_process() has
         * already woken them in this case.
         */
        if (!(qcc->flags & (QC_CF_ERR_CONN|QC_CF_ERRL)))
                qcc_wake_streams(qcc);

        qcc_refresh_timeout(qcc);

        TRACE_LEAVE(QMUX_EV_QCC_WAKE, conn);
        return 0;

 release:
        qcc_app_shutdown(qcc);
        qcc_release(qcc);
        TRACE_LEAVE(QMUX_EV_QCC_WAKE);
        return 1;
}

static void qcm_strm_shut(struct stconn *sc, unsigned int mode, struct se_abort_info *reason)
{
        struct qcs *qcs = __sc_mux_strm(sc);
        struct qcc *qcc = qcs->qcc;
        enum qcc_app_ops_lclose_mode lclo_mode;

        if (!(mode & (SE_SHW_SILENT|SE_SHW_NORMAL)))
                return;

        TRACE_ENTER(QMUX_EV_STRM_SHUT, qcc->conn, qcs);

        /* Not implemented. */
        BUG_ON(mode & SE_SHR_DRAIN);

        /* Early closure reported if QC_SF_FIN_STREAM not yet set. */
        if ((mode & (SE_SHW_SILENT|SE_SHW_NORMAL)) &&
            !qcs_is_close_local(qcs) &&
            !(qcs->flags & (QC_SF_FIN_STREAM|QC_SF_TO_RESET))) {

                if (qcs->flags & QC_SF_UNKNOWN_PL_LENGTH)
                        lclo_mode = QCC_APP_OPS_LCLO_MODE_NORMAL;
                else if (se_fl_test(qcs->sd, SE_FL_KILL_CONN))
                        lclo_mode = QCC_APP_OPS_LCLO_MODE_KILL_CONN;
                else
                        lclo_mode = QCC_APP_OPS_LCLO_MODE_ABORT;

                qcc->app_ops->lclose(qcs, lclo_mode);
                tasklet_wakeup(qcc->wait_event.tasklet);
        }

        if ((mode & SE_SHR_RESET) &&
            !qcs_is_close_remote(qcs) && !(qcs->flags & (QC_SF_READ_ABORTED))) {
                qcc->app_ops->lclose(qcs, QCC_APP_OPS_LCLO_MODE_READ);
        }

 out:
        TRACE_LEAVE(QMUX_EV_STRM_SHUT, qcc->conn, qcs);
}

static int qcm_ctl(struct connection *conn, enum mux_ctl_type mux_ctl, void *output)
{
        struct qcc *qcc = conn->ctx;

        switch (mux_ctl) {
        case MUX_CTL_EXIT_STATUS:
                return MUX_ES_UNKNOWN;

        case MUX_CTL_GET_GLITCHES:
                return qcc->glitches;

        case MUX_CTL_GET_NBSTRM:
                return qcc->nb_hreq;

        case MUX_CTL_GET_MAXSTRM:
                return qcc->lfctl.ms_bidi_rel;

        case MUX_CTL_TEVTS:
                return qcc->term_evts_log;

        default:
                return -1;
        }
}

static int qcm_sctl(struct stconn *sc, enum mux_sctl_type mux_sctl, void *output)
{
        int ret = 0;
        const struct qcs *qcs = __sc_mux_strm(sc);
        const struct qcc *qcc = qcs->qcc;
        const struct connection *conn = qcc->conn;
        union mux_sctl_dbg_str_ctx *dbg_ctx;
        struct buffer *buf;

        switch (mux_sctl) {
        case MUX_SCTL_SID:
                if (output)
                        *((int64_t *)output) = qcs->id;
                return ret;

        case MUX_SCTL_DBG_STR:
                dbg_ctx = output;
                buf = get_trash_chunk();

                if (dbg_ctx->arg.debug_flags & MUX_SCTL_DBG_STR_L_MUXS)
                        qcm_dump_qcs_info(buf, qcs);

                if (dbg_ctx->arg.debug_flags & MUX_SCTL_DBG_STR_L_MUXC)
                        qcm_dump_qcc_info(buf, qcc);

                if (dbg_ctx->arg.debug_flags & MUX_SCTL_DBG_STR_L_CONN) {
                        chunk_appendf(buf, " conn.flg=%#08x conn.err_code=%u conn.evts=%s",
                                      conn->flags, conn->err_code,
                                      tevt_evts2str(conn->term_evts_log));
                }

                if (dbg_ctx->arg.debug_flags & MUX_SCTL_DBG_STR_L_XPRT)
                        conn->xprt->dump_info(buf, conn);

                dbg_ctx->ret.buf = *buf;
                return ret;

        case MUX_SCTL_TEVTS:
                return qcs->sd->term_evts_log;

        default:
                return -1;
        }
}

/* for debugging with CLI's "show sess" command. May emit multiple lines, each
 * new one being prefixed with <pfx>, if <pfx> is not NULL, otherwise a single
 * line is used. Each field starts with a space so it's safe to print it after
 * existing fields.
 */
static int qcm_strm_show_sd(struct buffer *msg, struct sedesc *sd, const char *pfx)
{
        struct qcs *qcs = sd->se;
        struct qcc *qcc;
        int ret = 0;

        if (!qcs)
                return ret;

        chunk_appendf(msg, " qcs=%p .flg=%#x .id=%llu .st=%s .ctx=%p, .err=%#llx/%#llx",
                      qcs, qcs->flags, (ull)qcs->id, qcs_st_to_str(qcs->st), qcs->ctx,
                      (ull)qcs->rs_err, (ull)qcs->ss_err);

        if (pfx)
                chunk_appendf(msg, "\n%s", pfx);

        qcc = qcs->qcc;
        chunk_appendf(msg, " qcc=%p .flg=%#x .nbsc=%llu .nbhreq=%llu, .task=%p",
                      qcc, qcc->flags, (ull)qcc->nb_sc, (ull)qcc->nb_hreq, qcc->task);
        return ret;
}


static const struct mux_ops quic_ops = {
        .init        = qcm_init,
        .destroy     = qcm_destroy,
        .detach      = qcm_strm_detach,
        .rcv_buf     = qcm_strm_rcv_buf,
        .snd_buf     = qcm_strm_snd_buf,
        .nego_fastfwd = qcm_strm_nego_ff,
        .done_fastfwd = qcm_strm_done_ff,
        .resume_fastfwd = qcm_strm_resume_ff,
        .subscribe   = qcm_strm_subscribe,
        .unsubscribe = qcm_strm_unsubscribe,
        .wake        = qcm_wake,
        .avail_streams = qcm_avail_streams,
        .used_streams = qcm_used_streams,
        .takeover    = NULL,  /* QUIC takeover support not implemented yet */
        .attach      = qcm_strm_attach,
        .shut        = qcm_strm_shut,
        .ctl         = qcm_ctl,
        .sctl        = qcm_sctl,
        .show_sd     = qcm_strm_show_sd,
        .flags = MX_FL_HTX|MX_FL_NO_UPG|MX_FL_FRAMED,
        .name = "QUIC",
};

void qcc_show_quic(struct qcc *qcc)
{
        const struct quic_conn *qc = qcc->conn->handle.qc;
        struct eb64_node *node;

        chunk_appendf(&trash, "  qcc=0x%p flags=0x%x sc=%llu hreq=%llu bwnd=%llu/%llu\n",
                      qcc, qcc->flags, (ullong)qcc->nb_sc, (ullong)qcc->nb_hreq,
                      (ullong)qcc->tx.buf_in_flight, (ullong)qc->path->cwnd);

        if (qcc_is_pacing_active(qcc->conn)) {
                chunk_appendf(&trash, "  pacing int_sent=%d last_sent=%d\n",
                              qcc->tx.paced_sent_ctr,
                              qcc->tx.pacer.last_sent);
        }

        node = eb64_first(&qcc->streams_by_id);
        while (node) {
                struct qcs *qcs = eb64_entry(node, struct qcs, by_id);
                chunk_appendf(&trash, "    qcs=0x%p id=%llu flags=0x%x st=%s",
                              qcs, (ullong)qcs->id, qcs->flags,
                              qcs_st_to_str(qcs->st));

                if (!quic_stream_is_uni(qcs->id) || !quic_stream_is_local(qcc, qcs->id)) {
                        chunk_appendf(&trash, " rxb=%u(%u)", qcs->rx.data.bcnt, qcs->rx.data.bmax);
                        chunk_appendf(&trash, " rxoff=%llu", (ullong)qcs->rx.offset);
                }

                if (!quic_stream_is_uni(qcs->id) || !quic_stream_is_remote(qcc, qcs->id)) {
                        if (qcs->tx.stream)
                                bdata_ctr_print(&trash, &qcs->tx.stream->data, "txb=");
                        chunk_appendf(&trash, " txoff=%llu(%llu) msd=%llu",
                                      (ullong)qcs->tx.fc.off_real,
                                      (ullong)qcs->tx.fc.off_soft - (ullong)qcs->tx.fc.off_real,
                                      (ullong)qcs->tx.fc.limit);
                }
                chunk_appendf(&trash, "\n");
                node = eb64_next(node);
        }
}

static struct mux_proto_list mux_proto_quic =
  { .mux_proto = IST("quic"), .mode = PROTO_MODE_HTTP, .side = PROTO_SIDE_BOTH, .mux = &quic_ops };

INITCALL1(STG_REGISTER, register_mux_proto, &mux_proto_quic);

static const struct mux_ops qmux_ops = {
        .init        = qcm_init,
        .destroy     = qcm_destroy,
        .detach      = qcm_strm_detach,
        .rcv_buf     = qcm_strm_rcv_buf,
        .snd_buf     = qcm_strm_snd_buf,
        .nego_fastfwd = qcm_strm_nego_ff,
        .done_fastfwd = qcm_strm_done_ff,
        .resume_fastfwd = qcm_strm_resume_ff,
        .subscribe   = qcm_strm_subscribe,
        .unsubscribe = qcm_strm_unsubscribe,
        .wake        = qcm_wake,
        .avail_streams = qcm_avail_streams,
        .used_streams = qcm_used_streams,
        .takeover    = NULL,  /* QUIC takeover support not implemented yet */
        .attach      = qcm_strm_attach,
        .shut        = qcm_strm_shut,
        .ctl         = qcm_ctl,
        .sctl        = qcm_sctl,
        .show_sd     = qcm_strm_show_sd,
        .flags = MX_FL_HTX|MX_FL_NO_UPG|MX_FL_EXPERIMENTAL,
        .name = "QMUX",
};

static struct mux_proto_list mux_proto_qmux =
  { .mux_proto = IST("qmux"), .mode = PROTO_MODE_HTTP, .side = PROTO_SIDE_BOTH, .mux = &qmux_ops,
    .alpn = "\002h3", .init_xprt = XPRT_QMUX };

INITCALL1(STG_REGISTER, register_mux_proto, &mux_proto_qmux);