/* soup-message-io-http2.c

 *

 * Copyright 2021 Igalia S.L.

 *

 * This file is free software; you can redistribute it and/or modify it

 * under the terms of the GNU Lesser General Public License as

 * published by the Free Software Foundation; either version 2 of the

 * License, or (at your option) any later version.

 *

 * This file is distributed in the hope that it will be useful, but

 * WITHOUT ANY WARRANTY; without even the implied warranty of

 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU

 * Lesser General Public License for more details.

 *

 * You should have received a copy of the GNU Lesser General Public

 * License along with this program.  If not, see <http://www.gnu.org/licenses/>.

 *

 * SPDX-License-Identifier: LGPL-2.0-or-later

 */

#ifdef HAVE_CONFIG_H

#include "config.h"

#endif

#undef G_LOG_DOMAIN

#define G_LOG_DOMAIN "libsoup-http2"

#include <glib.h>

#include <glib/gi18n-lib.h>

#include "soup-client-message-io-http2.h"

#include "soup-body-input-stream.h"

#include "soup-message-metrics-private.h"

#include "soup-message-headers-private.h"

#include "soup-message-private.h"

#include "soup-message-io-source.h"

#include "soup-message-queue-item.h"

#include "content-sniffer/soup-content-sniffer-stream.h"

#include "soup-client-input-stream.h"

#include "soup-logger-private.h"

#include "soup-uri-utils-private.h"

#include "soup-http2-utils.h"

#include "content-decoder/soup-content-decoder.h"

#include "soup-body-input-stream-http2.h"

#define FRAME_HEADER_SIZE 9

typedef struct {

        SoupClientMessageIO iface;

        GThread *owner;

        gboolean async;

        GWeakRef conn;

        GIOStream *stream;

        GInputStream *istream;

        GOutputStream *ostream;

        guint64 connection_id;

        GError *error;

        GSource *read_source;

        GSource *write_source;

        GSource *write_idle_source;

        GHashTable *messages;

        GHashTable *closed_messages;

        GList *pending_io_messages;

        nghttp2_session *session;

        /* Owned by nghttp2 */

        guint8 *write_buffer;

        gssize write_buffer_size;

        gssize written_bytes;

        gboolean is_shutdown;

        GTask *close_task;

        gboolean session_terminated;

        gboolean goaway_sent;

        gboolean ever_used;

        guint in_callback;

} SoupClientMessageIOHTTP2;

typedef struct {

        SoupMessageQueueItem *item;

        SoupMessage *msg;

        SoupMessageMetrics *metrics;

        GInputStream *decoded_data_istream;

        GInputStream *body_istream;

        GTask *task;

        gboolean in_io_try_sniff_content;

        /* Request body */

        SoupLogger *logger;

        gssize request_body_bytes_to_write;

        /* Pollable data sources */

        GSource *data_source_poll;

        /* Non-pollable data sources */

        GByteArray *data_source_buffer;

        GError *data_source_error;

        gboolean data_source_eof;

        GCancellable *data_source_cancellable;

        GCancellable *data_source_message_cancellable;

        gulong data_source_cancellable_id;

        SoupClientMessageIOHTTP2 *io; /* Unowned */

        SoupMessageIOCompletionFn completion_cb;

        gpointer completion_data;

        SoupHTTP2IOState state;

        GError *error;

        uint32_t http2_error;

        gboolean paused;

        guint32 stream_id;

        gboolean can_be_restarted;

        gboolean expect_continue;

        GSource *check_status_idle_source;

} SoupHTTP2MessageData;

static void soup_client_message_io_http2_finished (SoupClientMessageIO *iface, SoupMessage *msg);

static ssize_t on_data_source_read_callback (nghttp2_session *session, int32_t stream_id, uint8_t *buf, size_t length, uint32_t *data_flags, nghttp2_data_source *source, void *user_data);

G_GNUC_PRINTF(3, 0)

static void

h2_debug (SoupClientMessageIOHTTP2   *io,

          SoupHTTP2MessageData       *data,

          const char                 *format,

          ...)

{

        va_list args;

        char *message;

        guint32 stream_id = 0;

        if (g_log_writer_default_would_drop (G_LOG_LEVEL_DEBUG, G_LOG_DOMAIN))

                return;

  va_start (args, format);

  message = g_strdup_vprintf (format, args);

  va_end (args);

        if (data)

                stream_id = data->stream_id;

        g_assert (io);

        g_log (G_LOG_DOMAIN, G_LOG_LEVEL_DEBUG, "[CLIENT] [C%" G_GUINT64_FORMAT "-S%u] [%s] %s", io->connection_id, stream_id, data ? soup_http2_io_state_to_string (data->state) : "-", message);

        g_free (message);

}

static SoupClientMessageIOHTTP2 *

get_io_data (SoupMessage *msg)

{

        return (SoupClientMessageIOHTTP2 *)soup_message_get_io_data (msg);

}

static int

get_data_io_priority (SoupHTTP2MessageData *data)

{

  if (!data->item->task)

    return G_PRIORITY_DEFAULT;

  return g_task_get_priority (data->item->task);

}

static void

set_error_for_data (SoupHTTP2MessageData *data,

                    GError               *error)

{

        h2_debug (data->io, data, "[SESSION] Error: %s", error->message);

        /* First error is probably the one we want. */

        if (!data->error)

                data->error = error;

        else

                g_error_free (error);

}

static void

set_http2_error_for_data (SoupHTTP2MessageData *data,

                          uint32_t              error_code)

{

        h2_debug (data->io, data, "[SESSION] Error: %s", nghttp2_http2_strerror (error_code));

        if (data->error)

                return;

        data->http2_error = error_code;

        data->error = g_error_new (G_IO_ERROR, G_IO_ERROR_FAILED,

                                   "HTTP/2 Error: %s", nghttp2_http2_strerror (error_code));

}

static void

set_io_error (SoupClientMessageIOHTTP2 *io,

              GError                   *error)

{

        h2_debug (io, NULL, "[SESSION] IO error: %s", error->message);

        if (!io->error)

                io->error = error;

        else

                g_error_free (error);

        if (io->close_task && !io->goaway_sent) {

                g_task_return_boolean (io->close_task, TRUE);

                g_clear_object (&io->close_task);

        }

}

static void

advance_state_from (SoupHTTP2MessageData *data,

                    SoupHTTP2IOState      from,

                    SoupHTTP2IOState      to)

{

        if (data->state != from) {

                g_warning ("Unexpected state changed %s -> %s, expected to be from %s",

                           soup_http2_io_state_to_string (data->state), soup_http2_io_state_to_string (to),

                           soup_http2_io_state_to_string (from));

        }

        /* State never goes backwards */

        if (to < data->state) {

                g_warning ("Unexpected state changed %s -> %s, expected %s -> %s\n",

                           soup_http2_io_state_to_string (data->state), soup_http2_io_state_to_string (to),

                           soup_http2_io_state_to_string (from), soup_http2_io_state_to_string (to));

                return;

        }

        h2_debug (data->io, data, "[SESSION] State %s -> %s",

                  soup_http2_io_state_to_string (data->state), soup_http2_io_state_to_string (to));

        data->state = to;

}

static gboolean

soup_http2_message_data_can_be_restarted (SoupHTTP2MessageData *data,

                                          GError               *error)

{

        if (data->can_be_restarted)

                return TRUE;

        return data->state < STATE_READ_DATA_START &&

                data->io->ever_used &&

                !g_error_matches (error, G_IO_ERROR, G_IO_ERROR_TIMED_OUT) &&

                !g_error_matches (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK) &&

                !g_error_matches (error, G_IO_ERROR, G_IO_ERROR_CANCELLED) &&

                error->domain != G_TLS_ERROR &&

                data->http2_error == NGHTTP2_NO_ERROR &&

                SOUP_METHOD_IS_IDEMPOTENT (soup_message_get_method (data->msg));

}

static void

soup_http2_message_data_destroy_check_status_idle_source (SoupHTTP2MessageData *data)

{

        if (!data->check_status_idle_source)

                return;

        g_source_destroy (data->check_status_idle_source);

        g_clear_pointer (&data->check_status_idle_source, g_source_unref);

}

static void

soup_http2_message_data_check_status (SoupHTTP2MessageData *data)

{

        SoupClientMessageIOHTTP2 *io = data->io;

        SoupMessage *msg = data->msg;

        GTask *task = data->task;

        GError *error = NULL;

        soup_http2_message_data_destroy_check_status_idle_source (data);

        if (!task)

                return;

        if (g_cancellable_set_error_if_cancelled (g_task_get_cancellable (task), &error)) {

                io->pending_io_messages = g_list_remove (io->pending_io_messages, data);

                data->task = NULL;

                soup_client_message_io_http2_finished ((SoupClientMessageIO *)io, msg);

                g_task_return_error (task, error);

                g_object_unref (task);

                return;

        }

        if (data->paused)

                return;

        if (io->error && !data->error)

                data->error = g_error_copy (io->error);

        if (data->error) {

                GError *error = g_steal_pointer (&data->error);

                if (soup_http2_message_data_can_be_restarted (data, error))

                        data->item->state = SOUP_MESSAGE_RESTARTING;

                else

                        soup_message_set_metrics_timestamp (data->msg, SOUP_MESSAGE_METRICS_RESPONSE_END);

                io->pending_io_messages = g_list_remove (io->pending_io_messages, data);

                data->task = NULL;

                soup_client_message_io_http2_finished ((SoupClientMessageIO *)io, msg);

                g_task_return_error (task, error);

                g_object_unref (task);

                return;

        }

        if (data->state == STATE_READ_DATA_START && !soup_message_has_content_sniffer (msg))

                advance_state_from (data, STATE_READ_DATA_START, STATE_READ_DATA);

        if (data->state < STATE_READ_DATA)

                return;

        io->pending_io_messages = g_list_remove (io->pending_io_messages, data);

        data->task = NULL;

        g_task_return_boolean (task, TRUE);

        g_object_unref (task);

}

static gboolean

check_status_idle_source_cb (SoupHTTP2MessageData *data)

{

        g_clear_pointer (&data->check_status_idle_source, g_source_unref);

        soup_http2_message_data_check_status (data);

        return G_SOURCE_REMOVE;

}

static void

soup_http2_message_data_check_status_in_idle (SoupHTTP2MessageData *data)

{

        if (data->check_status_idle_source)

                return;

        data->check_status_idle_source = g_idle_source_new ();

        g_source_set_static_name (data->check_status_idle_source, "Soup HTTP/2 check message data status");

        g_source_set_priority (data->check_status_idle_source, G_PRIORITY_DEFAULT);

        g_source_set_callback (data->check_status_idle_source, (GSourceFunc)check_status_idle_source_cb, data, NULL);

        g_source_attach (data->check_status_idle_source, g_main_context_get_thread_default ());

}

static gboolean

io_write (SoupClientMessageIOHTTP2 *io,

          gboolean                  blocking,

          GCancellable             *cancellable,

          GError                  **error)

{

        /* We must write all of nghttp2's buffer before we ask for more */

        if (io->written_bytes == io->write_buffer_size)

                io->write_buffer = NULL;

        if (io->write_buffer == NULL) {

                io->written_bytes = 0;

                g_warn_if_fail (io->in_callback == 0);

                io->write_buffer_size = nghttp2_session_mem_send (io->session, (const guint8**)&io->write_buffer);

                NGCHECK (io->write_buffer_size);

                if (io->write_buffer_size == 0) {

                        /* Done */

                        io->write_buffer = NULL;

                        return TRUE;

                }

        }

        gssize ret = g_pollable_stream_write (io->ostream,

                                              io->write_buffer + io->written_bytes,

                                              io->write_buffer_size - io->written_bytes,

                                              blocking, cancellable, error);

        if (ret < 0)

                return FALSE;

        io->written_bytes += ret;

        return TRUE;

}

static gboolean

io_write_ready (GObject                  *stream,

                SoupClientMessageIOHTTP2 *io)

{

        GError *error = NULL;

        if (io->error) {

                g_clear_pointer (&io->write_source, g_source_unref);

                return G_SOURCE_REMOVE;

        }

        while (!error && nghttp2_session_want_write (io->session))

                io_write (io, FALSE, NULL, &error);

        if (g_error_matches (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK)) {

                g_error_free (error);

                return G_SOURCE_CONTINUE;

        }

        if (error)

                set_io_error (io, error);

        g_clear_pointer (&io->write_source, g_source_unref);

        return G_SOURCE_REMOVE;

}

static gboolean io_write_idle_cb (SoupClientMessageIOHTTP2* io);

static void

io_try_write (SoupClientMessageIOHTTP2 *io,

              gboolean                  blocking)

{

        GError *error = NULL;

        if (io->write_source)

                return;

        if (io->in_callback) {

                if (blocking || !nghttp2_session_want_write (io->session))

                        return;

                if (io->write_idle_source)

                        return;

                io->write_idle_source = g_idle_source_new ();

                g_source_set_static_name (io->write_idle_source, "Soup HTTP/2 write idle source");

                /* Give write more priority than read */

                g_source_set_priority (io->write_idle_source, G_PRIORITY_DEFAULT - 1);

                g_source_set_callback (io->write_idle_source, (GSourceFunc)io_write_idle_cb, io, NULL);

                g_source_attach (io->write_idle_source, g_main_context_get_thread_default ());

                return;

        }

        if (io->write_idle_source) {

                g_source_destroy (io->write_idle_source);

                g_clear_pointer (&io->write_idle_source, g_source_unref);

        }

        while (!error && nghttp2_session_want_write (io->session))

                io_write (io, blocking, NULL, &error);

        if (!blocking && g_error_matches (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK)) {

                g_clear_error (&error);

                io->write_source = g_pollable_output_stream_create_source (G_POLLABLE_OUTPUT_STREAM (io->ostream), NULL);

                g_source_set_static_name (io->write_source, "Soup HTTP/2 write source");

                /* Give write more priority than read */

                g_source_set_priority (io->write_source, G_PRIORITY_DEFAULT - 1);

                g_source_set_callback (io->write_source, (GSourceFunc)io_write_ready, io, NULL);

                g_source_attach (io->write_source, g_main_context_get_thread_default ());

                return;

        }

        if (error)

                set_io_error (io, error);

}

static gboolean

io_write_idle_cb (SoupClientMessageIOHTTP2* io)

{

        g_clear_pointer (&io->write_idle_source, g_source_unref);

        io_try_write (io, FALSE);

        return G_SOURCE_REMOVE;

}

static gboolean

io_read (SoupClientMessageIOHTTP2  *io,

         gboolean                   blocking,

         GCancellable              *cancellable,

         GError                   **error)

{

        guint8 buffer[16384];

        gssize read;

        int ret;

        /* Always try to write before read, in case there's a pending reset stream after an error. */

        io_try_write (io, blocking);

        if ((read = g_pollable_stream_read (io->istream, buffer, sizeof (buffer),

                                            blocking, cancellable, error)) < 0)

            return FALSE;

        if (read == 0) {

                g_set_error_literal (error, G_IO_ERROR,

                                     G_IO_ERROR_PARTIAL_INPUT,

                                     _("Connection terminated unexpectedly"));

                return FALSE;

        }

        g_warn_if_fail (io->in_callback == 0);

        ret = nghttp2_session_mem_recv (io->session, buffer, read);

        NGCHECK (ret);

        return ret > 0;

}

static gboolean

io_read_ready (GObject                  *stream,

               SoupClientMessageIOHTTP2 *io)

{

        GError *error = NULL;

        gboolean progress = TRUE;

        SoupConnection *conn;

        if (io->error) {

                g_clear_pointer (&io->read_source, g_source_unref);

                return G_SOURCE_REMOVE;

        }

        /* Mark the connection as in use to make sure it's not disconnected while

         * processing pending messages, for example if a goaway is received.

         */

        conn = g_weak_ref_get (&io->conn);

        if (conn)

                soup_connection_set_in_use (conn, TRUE);

        while (progress && nghttp2_session_want_read (io->session))

                progress = io_read (io, FALSE, NULL, &error);

        if (g_error_matches (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK)) {

                g_error_free (error);

                if (conn) {

                        soup_connection_set_in_use (conn, FALSE);

                        g_object_unref (conn);

                }

                return G_SOURCE_CONTINUE;

        }

        io->is_shutdown = TRUE;

        if (error) {

                set_io_error (io, error);

                g_list_foreach (io->pending_io_messages,

                                (GFunc)soup_http2_message_data_check_status,

                                NULL);

        }

        g_clear_pointer (&io->read_source, g_source_unref);

        if (conn) {

                soup_connection_set_in_use (conn, FALSE);

                g_object_unref (conn);

        }

        return G_SOURCE_REMOVE;

}

static void

sniff_for_empty_response (SoupMessage *msg)

{

        if (soup_message_has_content_sniffer (msg)) {

                const char *content_type = soup_message_headers_get_content_type (soup_message_get_response_headers (msg), NULL);

                if (!content_type)

                     content_type = "text/plain";

                soup_message_content_sniffed (msg, content_type, NULL);

        }

}

static gboolean

message_has_content_length_zero (SoupMessage *msg)

{

        SoupMessageHeaders *headers = soup_message_get_response_headers (msg);

        if (soup_message_headers_get_encoding (headers) != SOUP_ENCODING_CONTENT_LENGTH)

                return FALSE;

        return soup_message_headers_get_content_length (headers) == 0;

}

static void

io_try_sniff_content (SoupHTTP2MessageData *data,

                      gboolean              blocking,

                      GCancellable         *cancellable)

{

        GError *error = NULL;

        /* This can re-enter in sync mode */

        if (data->in_io_try_sniff_content)

                return;

        /* Don't read the body while paused (e.g. waiting for a compression

         * dictionary to be resolved). Sniffing reads through the content

         * decoder, which must not be fed data before it is ready. The sniff is

         * retried from soup_client_message_io_http2_unpause(). */

        if (data->paused)

                return;

        if (message_has_content_length_zero (data->msg)) {

                sniff_for_empty_response (data->msg);

                h2_debug (data->io, data, "[DATA] Sniffed content (Content-Length was 0)");

                advance_state_from (data, STATE_READ_DATA_START, STATE_READ_DATA);

                if (data->item->async)

                        soup_http2_message_data_check_status_in_idle (data);

                return;

        }

        data->in_io_try_sniff_content = TRUE;

        if (soup_message_try_sniff_content (data->msg, data->decoded_data_istream, blocking, cancellable, &error)) {

                h2_debug (data->io, data, "[DATA] Sniffed content");

                advance_state_from (data, STATE_READ_DATA_START, STATE_READ_DATA);

                if (data->item->async)

                        soup_http2_message_data_check_status_in_idle (data);

        } else {

                h2_debug (data->io, data, "[DATA] Sniffer stream was not ready %s", error->message);

                g_clear_error (&error);

        }

        data->in_io_try_sniff_content = FALSE;

}

static void

soup_client_message_io_http2_terminate_session (SoupClientMessageIOHTTP2 *io)

{

        if (io->session_terminated)

                return;

        if (g_hash_table_size (io->messages) != 0)

                return;

        io->session_terminated = TRUE;

        NGCHECK (nghttp2_session_terminate_session (io->session, NGHTTP2_NO_ERROR));

        io_try_write (io, !io->async);

}

/* HTTP2 read callbacks */

static int

on_header_callback (nghttp2_session     *session,

                    const nghttp2_frame *frame,

                    const uint8_t       *name,

                    size_t               namelen,

                    const uint8_t       *value,

                    size_t               valuelen,

                    uint8_t              flags,

                    void                *user_data)

{

        SoupHTTP2MessageData *data = nghttp2_session_get_stream_user_data (session, frame->hd.stream_id);

        if (!data)

                return 0;

        data->io->in_callback++;

        SoupMessage *msg = data->msg;

        if (name[0] == ':') {

                if (strcmp ((char *)name, ":status") == 0) {

                        guint status_code = (guint)g_ascii_strtoull ((char *)value, NULL, 10);

                        soup_message_set_status (msg, status_code, NULL);

                        data->io->in_callback--;

                        return 0;

                }

                g_debug ("Unknown header: %s = %s", name, value);

                data->io->in_callback--;

                return 0;

        }

        soup_message_headers_append_untrusted_data (soup_message_get_response_headers (data->msg),

                                                    (const char*)name, (const char*)value);

        data->io->in_callback--;

        return 0;

}

static int

on_invalid_header_callback (nghttp2_session     *session,

                            const nghttp2_frame *frame,

                            const uint8_t       *name,

                            size_t               namelen,

                            const uint8_t       *value,

                            size_t               valuelen,

                            uint8_t              flags,

                            void                *user_data)

{

        SoupHTTP2MessageData *data = nghttp2_session_get_stream_user_data (session, frame->hd.stream_id);

        h2_debug (user_data, data, "[HEADERS] Invalid header received: name=[%.*s] value=[%.*s]", namelen, name, valuelen, value);

        return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE;

}

static GError *

memory_stream_need_more_data_callback (SoupBodyInputStreamHttp2 *stream,

                                       GCancellable             *cancellable,

                                       gpointer                  user_data)

{

        SoupHTTP2MessageData *data = (SoupHTTP2MessageData*)user_data;

        GError *error = NULL;

        if (data->in_io_try_sniff_content)

                return NULL;

        if (nghttp2_session_want_read (data->io->session))

                io_read (data->io, TRUE, cancellable, &error);

        return error;

}

static void

memory_stream_read_data (SoupBodyInputStreamHttp2 *stream,

                         guint64                   bytes_read,

                         gpointer                  user_data)

{

        SoupHTTP2MessageData *data = (SoupHTTP2MessageData*)user_data;

        h2_debug (data->io, data, "[BODY_STREAM] Consumed %" G_GUINT64_FORMAT " bytes", bytes_read);

        NGCHECK (nghttp2_session_consume(data->io->session, data->stream_id, (size_t)bytes_read));

        io_try_write (data->io, !data->item->async);

}

static void

soup_http2_message_data_consume_buffered_body (SoupHTTP2MessageData *data)

{

        gsize buffered;

        gssize skipped;

        GError *error = NULL;

        if (!data->body_istream)

                return;

        buffered = soup_body_input_stream_http2_get_buffer_size (SOUP_BODY_INPUT_STREAM_HTTP2 (data->body_istream));

        if (!buffered)

                return;

        skipped = g_input_stream_skip (data->body_istream, buffered, NULL, &error);

        if (skipped < 0) {

                h2_debug (data->io, data, "[BODY_STREAM] Failed to consume %" G_GSIZE_FORMAT " buffered bytes before cancel: %s",

                          buffered, error->message);

                g_clear_error (&error);

                return;

        }

        h2_debug (data->io, data, "[BODY_STREAM] Consumed %zd/%" G_GSIZE_FORMAT " buffered bytes before cancel", skipped, buffered);

}

static int

on_begin_frame_callback (nghttp2_session        *session,

                         const nghttp2_frame_hd *hd,

                         void                   *user_data)

{

        SoupHTTP2MessageData *data = nghttp2_session_get_stream_user_data (session, hd->stream_id);

        h2_debug (user_data, data, "[RECV] [%s] Beginning: stream_id=%u", soup_http2_frame_type_to_string (hd->type), hd->stream_id);

        if (!data)

                return 0;

        data->io->in_callback++;

        switch (hd->type) {

        case NGHTTP2_HEADERS:

                if (data->state == STATE_WRITE_DONE) {

                        soup_message_set_metrics_timestamp (data->item->msg, SOUP_MESSAGE_METRICS_RESPONSE_START);

                        advance_state_from (data, STATE_WRITE_DONE, STATE_READ_HEADERS);

                }

                break;

        case NGHTTP2_DATA:

                if (data->state < STATE_READ_DATA_START) {

                        g_assert (!data->body_istream);

                        data->body_istream = soup_body_input_stream_http2_new ();

                        g_signal_connect (data->body_istream, "need-more-data",

                                          G_CALLBACK (memory_stream_need_more_data_callback), data);

                        g_signal_connect (data->body_istream, "read-data",

                                          G_CALLBACK (memory_stream_read_data), data);

                        g_assert (!data->decoded_data_istream);

                        data->decoded_data_istream = soup_session_setup_message_body_input_stream (data->item->session,

                                                                                                   data->msg,

                                                                                                   data->body_istream,

                                                                                                   SOUP_STAGE_MESSAGE_BODY);

                        advance_state_from (data, STATE_READ_HEADERS, STATE_READ_DATA_START);

                }

                break;

        }

        data->io->in_callback--;

        return 0;

}

static void

handle_goaway (SoupClientMessageIOHTTP2 *io,

               guint32                   error_code,

               int32_t                   last_stream_id)

{

        GHashTableIter iter;

        SoupHTTP2MessageData *data;

        if (last_stream_id == G_MAXINT32)

                return;

        g_hash_table_iter_init (&iter, io->messages);

        while (g_hash_table_iter_next (&iter, NULL, (gpointer*)&data)) {

                if ((int32_t)data->stream_id > last_stream_id) {

                        /* RFC-9113 §6.8: stream was not processed by the server,

                         * it can be retried on a new connection regardless of error_code. */

                        data->can_be_restarted = TRUE;

                        set_http2_error_for_data (data, error_code);

                } else if (error_code != NGHTTP2_NO_ERROR) {

                        /* Stream might have been processed by the server but connection is dying with an error.

                         * Fail explicitly to avoid hanging indefinitely if the server is slow or fails

                         * to close the TCP connection (which it MUST do per §5.4.1).

                         * Per §6.8 only idempotent methods may be retried for processed streams. */

                        if (SOUP_METHOD_IS_IDEMPOTENT (soup_message_get_method (data->msg)))

                                data->can_be_restarted = TRUE;

                        set_http2_error_for_data (data, error_code);

                }

                 /* else: Graceful shutdown (NO_ERROR) and the server might have processed this stream.

                  * Per §6.8 it might still complete successfully, let it finish normally. */

        }

}

static int

on_frame_recv_callback (nghttp2_session     *session,

                        const nghttp2_frame *frame,

                        gpointer             user_data)

{

        SoupClientMessageIOHTTP2 *io = user_data;

        SoupHTTP2MessageData *data;

        io->in_callback++;

        if (frame->hd.stream_id == 0) {

                h2_debug (io, NULL, "[RECV] [%s] Received: stream_id=%u, flags=%u", soup_http2_frame_type_to_string (frame->hd.type), frame->hd.stream_id, frame->hd.flags);

                switch (frame->hd.type) {

                case NGHTTP2_GOAWAY:

                        h2_debug (io, NULL, "[RECV] GOAWAY: error=%s, last_stream_id=%d %s",

                                  nghttp2_http2_strerror (frame->goaway.error_code),

                                  frame->goaway.last_stream_id,

                                  frame->goaway.opaque_data ? (char *)frame->goaway.opaque_data : "");

                        handle_goaway (io, frame->goaway.error_code, frame->goaway.last_stream_id);

                        io->is_shutdown = TRUE;

                        soup_client_message_io_http2_terminate_session (io);

                        break;

                case NGHTTP2_WINDOW_UPDATE:

                        h2_debug (io, NULL, "[RECV] WINDOW_UPDATE: increment=%d, total=%d", frame->window_update.window_size_increment,

                                  nghttp2_session_get_remote_window_size (session));

                        break;

                }

                io->in_callback--;

                return 0;

        }

        data = nghttp2_session_get_stream_user_data (session, frame->hd.stream_id);

        h2_debug (io, data, "[RECV] [%s] Received: stream_id=%u, flags=%u", soup_http2_frame_type_to_string (frame->hd.type), frame->hd.stream_id, frame->hd.flags);

        if (!data) {

                /* This can happen in case of cancellation */

                io->in_callback--;

                return 0;

        }

        switch (frame->hd.type) {

        case NGHTTP2_HEADERS: {

                guint status = soup_message_get_status (data->msg);

                if (data->metrics)

                        data->metrics->response_header_bytes_received += frame->hd.length + FRAME_HEADER_SIZE;

                h2_debug (io, data, "[HEADERS] category=%s status=%u",

                          soup_http2_headers_category_to_string (frame->headers.cat), status);

                switch (frame->headers.cat) {

                case NGHTTP2_HCAT_HEADERS:

                        if (!(frame->hd.flags & NGHTTP2_FLAG_END_HEADERS)) {

                                io->in_callback--;

                                return 0;

                        }

                        break;

                case NGHTTP2_HCAT_RESPONSE:

                        if (SOUP_STATUS_IS_INFORMATIONAL (status)) {

                                if (data->expect_continue && status == SOUP_STATUS_CONTINUE) {

                                        nghttp2_data_provider data_provider;

                                        data_provider.source.ptr = soup_message_get_request_body_stream (data->msg);

                                        data_provider.read_callback = on_data_source_read_callback;

                                        goffset content_length = soup_message_headers_get_content_length (soup_message_get_request_headers (data->msg));

                                        data->request_body_bytes_to_write = content_length > 0 ? content_length : -1;

                                        nghttp2_submit_data (io->session, NGHTTP2_FLAG_END_STREAM, frame->hd.stream_id, &data_provider);

                                        io_try_write (io, !data->item->async);

                                }

                                soup_message_got_informational (data->msg);

                                soup_message_cleanup_response (data->msg);

                                io->in_callback--;

                                return 0;

                        }

                        break;

                case NGHTTP2_HCAT_PUSH_RESPONSE:

                        g_warn_if_reached ();

                        break;

                default:

                        g_assert_not_reached ();

                }

                soup_message_got_headers (data->msg);

                if (soup_message_get_status (data->msg) == SOUP_STATUS_NO_CONTENT || frame->hd.flags & NGHTTP2_FLAG_END_STREAM) {

                        h2_debug (io, data, "Stream done");

                        advance_state_from (data, STATE_READ_HEADERS, STATE_READ_DATA_START);

                        sniff_for_empty_response (data->msg);

                        advance_state_from (data, STATE_READ_DATA_START, STATE_READ_DATA);

                        if (data->item->async)

                                soup_http2_message_data_check_status_in_idle (data);

                }

                break;

        }

        case NGHTTP2_DATA:

                h2_debug (io, data, "[RECV] [DATA] window=%d/%d", nghttp2_session_get_stream_effective_recv_data_length (session, frame->hd.stream_id),

                          nghttp2_session_get_stream_effective_local_window_size (session, frame->hd.stream_id));

                if (data->metrics)

                        data->metrics->response_body_bytes_received += frame->data.hd.length + FRAME_HEADER_SIZE;

                soup_message_got_body_data (data->msg, frame->data.hd.length + FRAME_HEADER_SIZE);

                if (frame->hd.flags & NGHTTP2_FLAG_END_STREAM) {

                        if (data->body_istream) {