Merge branch 'awslabs:main' into main

quinnj · web-flow · commit 3c4527eb0516 · 2025-04-15T22:07:12.000-06:00
diff --git a/include/aws/http/exports.h b/include/aws/http/exports.h
@@ -6,7 +6,7 @@
  * SPDX-License-Identifier: Apache-2.0.
  */
 
-#if defined(USE_WINDOWS_DLL_SEMANTICS) || defined(WIN32)
+#if defined(AWS_CRT_USE_WINDOWS_DLL_SEMANTICS) || defined(_WIN32)
 #    ifdef AWS_HTTP_USE_IMPORT_EXPORT
 #        ifdef AWS_HTTP_EXPORTS
 #            define AWS_HTTP_API __declspec(dllexport)
@@ -18,12 +18,12 @@
 #    endif /* USE_IMPORT_EXPORT */
 
 #else
-#    if ((__GNUC__ >= 4) || defined(__clang__)) && defined(AWS_HTTP_USE_IMPORT_EXPORT) && defined(AWS_HTTP_EXPORTS)
+#    if defined(AWS_HTTP_USE_IMPORT_EXPORT) && defined(AWS_HTTP_EXPORTS)
 #        define AWS_HTTP_API __attribute__((visibility("default")))
 #    else
 #        define AWS_HTTP_API
-#    endif /* __GNUC__ >= 4 || defined(__clang__) */
+#    endif
 
-#endif /* defined(USE_WINDOWS_DLL_SEMANTICS) || defined(WIN32) */
+#endif /* defined(AWS_CRT_USE_WINDOWS_DLL_SEMANTICS) || defined(_WIN32) */
 
 #endif /* AWS_HTTP_EXPORTS_H */
diff --git a/include/aws/http/private/h2_frames.h b/include/aws/http/private/h2_frames.h
@@ -207,6 +207,7 @@ int aws_h2_encode_frame(
  * body_complete will be set true if encoder reaches the end of the body_stream.
  * body_stalled will be true if aws_input_stream_read() stopped early (didn't
  * complete, though more space was available).
+ * body_failed will be true if the aws_input_stream was the cause of an error.
  *
  * Each call to this function encodes a complete DATA frame, or nothing at all,
  * so it's always safe to encode a different frame type or the body of a different stream
@@ -223,7 +224,8 @@ int aws_h2_encode_data_frame(
     size_t *connection_window_size_peer,
     struct aws_byte_buf *output,
     bool *body_complete,
-    bool *body_stalled);
+    bool *body_stalled,
+    bool *body_failed);
 
 AWS_HTTP_API
 void aws_h2_frame_destroy(struct aws_h2_frame *frame);
diff --git a/source/h2_frames.c b/source/h2_frames.c
@@ -321,13 +321,15 @@ int aws_h2_encode_data_frame(
     size_t *connection_window_size_peer,
     struct aws_byte_buf *output,
     bool *body_complete,
-    bool *body_stalled) {
+    bool *body_stalled,
+    bool *body_failed) {
 
     AWS_PRECONDITION(encoder);
     AWS_PRECONDITION(body_stream);
     AWS_PRECONDITION(output);
     AWS_PRECONDITION(body_complete);
     AWS_PRECONDITION(body_stalled);
+    AWS_PRECONDITION(body_failed);
     AWS_PRECONDITION(*stream_window_size_peer > 0);
 
     if (aws_h2_validate_stream_id(stream_id)) {
@@ -336,6 +338,7 @@ int aws_h2_encode_data_frame(
 
     *body_complete = false;
     *body_stalled = false;
+    *body_failed = false;
     uint8_t flags = 0;
 
     /*
@@ -378,12 +381,14 @@ int aws_h2_encode_data_frame(
 
     /* Read body into sub-buffer */
     if (aws_input_stream_read(body_stream, &body_sub_buf)) {
+        *body_failed = true;
         goto error;
     }
 
     /* Check if we've reached the end of the body */
     struct aws_stream_status body_status;
     if (aws_input_stream_get_status(body_stream, &body_status)) {
+        *body_failed = true;
         goto error;
     }
 
diff --git a/source/h2_stream.c b/source/h2_stream.c
@@ -407,6 +407,7 @@ static void s_stream_data_write_destroy(
 
     AWS_PRECONDITION(stream);
     AWS_PRECONDITION(write);
+    AWS_PRECONDITION(!aws_linked_list_node_is_in_list(&stream->node));
     if (write->on_complete) {
         write->on_complete(&stream->base, error_code, write->user_data);
     }
@@ -682,37 +683,13 @@ static inline bool s_h2_stream_has_outgoing_writes(struct aws_h2_stream *stream)
     return !aws_linked_list_empty(&stream->thread_data.outgoing_writes);
 }
 
-static void s_h2_stream_write_data_complete(struct aws_h2_stream *stream, bool *waiting_writes) {
-    AWS_PRECONDITION(waiting_writes);
-    AWS_PRECONDITION(s_h2_stream_has_outgoing_writes(stream));
-
-    /* finish/clean up the current write operation */
-    struct aws_linked_list_node *node = aws_linked_list_pop_front(&stream->thread_data.outgoing_writes);
-    struct aws_h2_stream_data_write *write_op = AWS_CONTAINER_OF(node, struct aws_h2_stream_data_write, node);
-    const bool ending_stream = write_op->end_stream;
-    s_stream_data_write_destroy(stream, write_op, AWS_OP_SUCCESS);
-
-    /* check to see if there are more queued writes or stream_end was called */
-    *waiting_writes = !ending_stream && !s_h2_stream_has_outgoing_writes(stream);
-}
-
 static struct aws_h2_stream_data_write *s_h2_stream_get_current_write(struct aws_h2_stream *stream) {
     AWS_PRECONDITION(s_h2_stream_has_outgoing_writes(stream));
     struct aws_linked_list_node *node = aws_linked_list_front(&stream->thread_data.outgoing_writes);
     struct aws_h2_stream_data_write *write = AWS_CONTAINER_OF(node, struct aws_h2_stream_data_write, node);
     return write;
 }
 
-static struct aws_input_stream *s_h2_stream_get_data_stream(struct aws_h2_stream *stream) {
-    struct aws_h2_stream_data_write *write = s_h2_stream_get_current_write(stream);
-    return write->data_stream;
-}
-
-static bool s_h2_stream_does_current_write_end_stream(struct aws_h2_stream *stream) {
-    struct aws_h2_stream_data_write *write = s_h2_stream_get_current_write(stream);
-    return write->end_stream;
-}
-
 int aws_h2_stream_on_activated(struct aws_h2_stream *stream, enum aws_h2_stream_body_state *body_state) {
     AWS_PRECONDITION_ON_CHANNEL_THREAD(stream);
 
@@ -799,12 +776,14 @@ int aws_h2_stream_encode_data_frame(
     }
 
     *data_encode_status = AWS_H2_DATA_ENCODE_COMPLETE;
-    struct aws_input_stream *input_stream = s_h2_stream_get_data_stream(stream);
+    struct aws_h2_stream_data_write *current_write = s_h2_stream_get_current_write(stream);
+    struct aws_input_stream *input_stream = current_write->data_stream;
     AWS_ASSERT(input_stream);
 
     bool input_stream_complete = false;
     bool input_stream_stalled = false;
-    bool ends_stream = s_h2_stream_does_current_write_end_stream(stream);
+    bool input_stream_failed = false;
+    bool ends_stream = current_write->end_stream;
     if (aws_h2_encode_data_frame(
             encoder,
             stream->base.id,
@@ -815,20 +794,30 @@ int aws_h2_stream_encode_data_frame(
             &connection->thread_data.window_size_peer,
             output,
             &input_stream_complete,
-            &input_stream_stalled)) {
+            &input_stream_stalled,
+            &input_stream_failed)) {
+
+        int error_code = aws_last_error();
+
+        /* If error cause caused aws_input_stream, report that specific error in its write-completion callback */
+        if (input_stream_failed) {
+            aws_linked_list_remove(&current_write->node);
+            s_stream_data_write_destroy(stream, current_write, error_code);
+        }
 
         /* Failed to write DATA, treat it as a Stream Error */
-        AWS_H2_STREAM_LOGF(ERROR, stream, "Error encoding stream DATA, %s", aws_error_name(aws_last_error()));
-        struct aws_h2err returned_h2err = s_send_rst_and_close_stream(stream, aws_h2err_from_last_error());
+        AWS_H2_STREAM_LOGF(ERROR, stream, "Error encoding stream DATA, %s", aws_error_name(error_code));
+        struct aws_h2err returned_h2err = s_send_rst_and_close_stream(stream, aws_h2err_from_aws_code(error_code));
         if (aws_h2err_failed(returned_h2err)) {
             aws_h2_connection_shutdown_due_to_write_err(connection, returned_h2err.aws_code);
         }
         return AWS_OP_SUCCESS;
     }
 
-    bool waiting_writes = false;
     if (input_stream_complete) {
-        s_h2_stream_write_data_complete(stream, &waiting_writes);
+        /* finish/clean up the current write operation */
+        aws_linked_list_remove(&current_write->node);
+        s_stream_data_write_destroy(stream, current_write, AWS_ERROR_SUCCESS);
     }
 
     /*
@@ -867,10 +856,11 @@ int aws_h2_stream_encode_data_frame(
              * from outgoing list */
             *data_encode_status = AWS_H2_DATA_ENCODE_ONGOING_WINDOW_STALLED;
         }
-        if (waiting_writes) {
+        if (!s_h2_stream_has_outgoing_writes(stream)) {
             /* if window stalled and we waiting for manual writes, we take waiting writes status, which will be handled
              * properly if more writes coming, but windows is still stalled. But not the other way around. */
             AWS_ASSERT(input_stream_complete);
+            AWS_ASSERT(!ends_stream);
             *data_encode_status = AWS_H2_DATA_ENCODE_ONGOING_WAITING_FOR_WRITES;
         }
     }
diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
@@ -502,6 +502,7 @@ add_test_case(h2_client_error_from_incoming_headers_callback_reset_stream)
 add_test_case(h2_client_error_from_incoming_headers_done_callback_reset_stream)
 add_test_case(h2_client_error_from_incoming_body_callback_reset_stream)
 add_test_case(h2_client_manual_data_write)
+add_test_case(h2_client_manual_data_write_read_broken)
 add_test_case(h2_client_manual_data_write_not_enabled)
 add_test_case(h2_client_manual_data_write_with_body)
 add_test_case(h2_client_manual_data_write_no_data)
diff --git a/tests/fuzz/fuzz_h2_decoder_correct.c b/tests/fuzz/fuzz_h2_decoder_correct.c
@@ -236,6 +236,7 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) {
 
             bool body_complete;
             bool body_stalled;
+            bool body_failed;
             int32_t stream_window_size_peer = AWS_H2_WINDOW_UPDATE_MAX;
             size_t connection_window_size_peer = AWS_H2_WINDOW_UPDATE_MAX;
             AWS_FATAL_ASSERT(
@@ -249,7 +250,8 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) {
                     &connection_window_size_peer,
                     &frame_data,
                     &body_complete,
-                    &body_stalled) == AWS_OP_SUCCESS);
+                    &body_stalled,
+                    &body_failed) == AWS_OP_SUCCESS);
 
             struct aws_stream_status body_status;
             aws_input_stream_get_status(body, &body_status);
diff --git a/tests/h2_test_helper.c b/tests/h2_test_helper.c
@@ -594,6 +594,7 @@ int h2_fake_peer_send_data_frame_with_padding_length(
 
     bool body_complete;
     bool body_stalled;
+    bool body_failed;
     int32_t stream_window_size_peer = AWS_H2_WINDOW_UPDATE_MAX;
     size_t connection_window_size_peer = AWS_H2_WINDOW_UPDATE_MAX;
     ASSERT_SUCCESS(aws_h2_encode_data_frame(
@@ -606,10 +607,12 @@ int h2_fake_peer_send_data_frame_with_padding_length(
         &connection_window_size_peer,
         &msg->message_data,
         &body_complete,
-        &body_stalled));
+        &body_stalled,
+        &body_failed));
 
     ASSERT_TRUE(body_complete);
     ASSERT_FALSE(body_stalled);
+    ASSERT_FALSE(body_failed);
     ASSERT_TRUE(msg->message_data.len != 0);
 
     ASSERT_SUCCESS(testing_channel_push_read_message(peer->testing_channel, msg));
diff --git a/tests/py_localhost/server.py b/tests/py_localhost/server.py
@@ -308,7 +308,9 @@ def window_updated(self, stream_id, delta):
 ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
 ssl_context.options |= (ssl.OP_NO_COMPRESSION)
 ssl_context.load_cert_chain(
-    certfile="../resources/unittests.crt", keyfile="../resources/unittests.key")
+    certfile=os.path.join(os.path.dirname(__file__),
+                          "../resources/unittests.crt"),
+    keyfile=os.path.join(os.path.dirname(__file__), "../resources/unittests.key"))
 ssl_context.set_alpn_protocols(["h2"])
 
 loop = asyncio.new_event_loop()
diff --git a/tests/stream_test_helper.c b/tests/stream_test_helper.c
@@ -201,6 +201,7 @@ int client_stream_tester_init(
         .on_metrics = s_on_metrics,
         .on_complete = s_on_complete,
         .on_destroy = s_on_destroy,
+        .http2_use_manual_data_writes = options->http2_manual_write,
     };
     tester->stream = aws_http_connection_make_request(options->connection, &request_options);
     ASSERT_NOT_NULL(tester->stream);
diff --git a/tests/stream_test_helper.h b/tests/stream_test_helper.h
@@ -51,6 +51,7 @@ struct client_stream_tester {
 struct client_stream_tester_options {
     struct aws_http_message *request;
     struct aws_http_connection *connection;
+    bool http2_manual_write;
 };
 
 int client_stream_tester_init(
diff --git a/tests/test_h2_client.c b/tests/test_h2_client.c
@@ -5422,6 +5422,70 @@ TEST_CASE(h2_client_manual_data_write) {
     return s_tester_clean_up();
 }
 
+static void s_http_stream_write_complete_fn(struct aws_http_stream *stream, int error_code, void *user_data) {
+    (void)stream;
+    int *ctx = (int *)user_data;
+    *ctx = error_code;
+}
+
+TEST_CASE(h2_client_manual_data_write_read_broken) {
+
+    ASSERT_SUCCESS(s_tester_init(allocator, ctx));
+    /* get connection preface and acks out of the way */
+    ASSERT_SUCCESS(h2_fake_peer_send_connection_preface_default_settings(&s_tester.peer));
+    ASSERT_SUCCESS(h2_fake_peer_decode_messages_from_testing_channel(&s_tester.peer));
+
+    struct aws_http_message *request = aws_http2_message_new_request(allocator);
+    ASSERT_NOT_NULL(request);
+
+    struct aws_http_header request_headers_src[] = {
+        DEFINE_HEADER(":method", "POST"),
+        DEFINE_HEADER(":scheme", "https"),
+        DEFINE_HEADER(":path", "/"),
+    };
+    aws_http_message_add_header_array(request, request_headers_src, AWS_ARRAY_SIZE(request_headers_src));
+    struct client_stream_tester stream_tester;
+
+    struct client_stream_tester_options options = {
+        .request = request,
+        .connection = s_tester.connection,
+        .http2_manual_write = true,
+    };
+    ASSERT_SUCCESS(client_stream_tester_init(&stream_tester, s_tester.alloc, &options));
+    struct aws_http_stream *stream = stream_tester.stream;
+    ASSERT_NOT_NULL(stream);
+
+    testing_channel_drain_queued_tasks(&s_tester.testing_channel);
+
+    struct aws_input_stream *data_stream = aws_input_stream_new_tester(allocator, aws_byte_cursor_from_c_str("abcd"));
+    aws_input_stream_tester_set_reading_broken(data_stream, true);
+    int error_code = 0;
+    struct aws_http2_stream_write_data_options write = {
+        .data = data_stream,
+        .on_complete = s_http_stream_write_complete_fn,
+        .user_data = &error_code,
+    };
+
+    ASSERT_SUCCESS(aws_http2_stream_write_data(stream, &write));
+    aws_input_stream_release(data_stream);
+
+    testing_channel_drain_queued_tasks(&s_tester.testing_channel);
+    ASSERT_TRUE(stream_tester.complete);
+    /* The stream complete will get the error code from the input stream read. */
+    ASSERT_UINT_EQUALS(stream_tester.on_complete_error_code, AWS_IO_STREAM_READ_FAILED);
+    /* The write triggers the error, which should be reported to the write complete */
+    ASSERT_UINT_EQUALS(error_code, AWS_IO_STREAM_READ_FAILED);
+
+    aws_http_message_release(request);
+
+    /* close the connection */
+    aws_http_connection_close(s_tester.connection);
+    client_stream_tester_clean_up(&stream_tester);
+
+    /* clean up */
+    return s_tester_clean_up();
+}
+
 TEST_CASE(h2_client_manual_data_write_not_enabled) {
 
     ASSERT_SUCCESS(s_tester_init(allocator, ctx));
diff --git a/tests/test_h2_encoder.c b/tests/test_h2_encoder.c
