why we try to finalize the checksum on error?

Author: TingDaoK

source/s3_checksum_stream.c

@@ -113,7 +113,7 @@ static void s_aws_input_checksum_stream_destroy(struct aws_checksum_stream *impl
     }
 
     /* Compute the checksum of whatever was read, if we didn't reach the end of the underlying stream. */
-    s_finalize_checksum(impl);
+    // s_finalize_checksum(impl);
 
     aws_checksum_destroy(impl->checksum);
     aws_input_stream_release(impl->old_stream);

tests/CMakeLists.txt

@@ -331,6 +331,7 @@ if(ENABLE_MOCK_SERVER_TESTS)
     add_net_test_case(multipart_upload_unsigned_with_trailer_checksum_mock_server)
     add_net_test_case(single_upload_unsigned_with_trailer_checksum_mock_server)
     add_net_test_case(multipart_upload_with_network_interface_names_mock_server)
+    add_net_test_case(multipart_upload_checksum_with_retry_before_finish_mock_server)
     add_net_test_case(multipart_upload_checksum_with_retry_mock_server)
     add_net_test_case(multipart_download_checksum_with_retry_mock_server)
     add_net_test_case(async_internal_error_from_complete_multipart_mock_server)

tests/mock_s3_server/UploadPart/throttle_before_finish.json

@@ -0,0 +1,13 @@
+{
+    "status": 503,
+    "headers": {"ETag": "b54357faf0632cce46e942fa68356b38", "Connection": "keep-alive"},
+    "body": [
+        "<?xml version=\"1.0\" encoding=\"UTF-8\"?>",
+        "",
+        "<Error>",
+         "<Code>SlowDown</Code>",
+         "<RequestId>656c76696e6727732072657175657374</RequestId>",
+         "<HostId>Uuag1LuByRx9e6j5Onimru9pO4ZVKnJ2Qz7/C1NPcfTWAtRPfTaOFg==</HostId>",
+        "</Error>"
+    ]
+  }

tests/mock_s3_server/mock_s3_server.py

@@ -22,6 +22,7 @@
 
 # Flags to keep between requests
 SHOULD_THROTTLE = True
+SHOULD_SKIP_WAIT = True
 RETRY_REQUEST_COUNT = 0
 
 
@@ -63,6 +64,7 @@ def _resolve_file_path(self, wrapper, request_type):
         if self.json_path is None:
             response_file = os.path.join(
                 base_dir, request_type.name, f"{self.path[1:]}.json")
+            print(response_file)
             if os.path.exists(response_file) == False:
                 wrapper.info(
                     response_file, "not exist, using the default response")
@@ -75,6 +77,7 @@ def _resolve_file_path(self, wrapper, request_type):
                     response_file = os.path.join(
                         base_dir, request_type.name, f"default.json")
                 else:
+                    print("throttle")
                     wrapper.info("Throttling")
                 # Flip the flag
                 SHOULD_THROTTLE = not SHOULD_THROTTLE
@@ -480,12 +483,19 @@ async def handle_mock_s3_request(wrapper, request):
         # TODO: support more type.
         wrapper.info("unsupported request:", request)
         request_type = S3Opts.CreateMultipartUpload
-
-    while True:
-        event = await wrapper.next_event()
-        if type(event) is h11.EndOfMessage:
-            break
-        assert type(event) is h11.Data
+    global SHOULD_SKIP_WAIT
+    if "before_finish" not in parsed_path.path or request_type is not S3Opts.UploadPart:
+        # restore the state
+        SHOULD_SKIP_WAIT = True
+    if "before_finish" in parsed_path.path and SHOULD_SKIP_WAIT and request_type is S3Opts.UploadPart:
+        print("not wait for the request to complete")
+        SHOULD_SKIP_WAIT = False
+    else:
+        while True:
+            event = await wrapper.next_event()
+            if type(event) is h11.EndOfMessage:
+                break
+            assert type(event) is h11.Data
 
     if response_config is None:
         response_config = ResponseConfig(parsed_path.path)

tests/s3_mock_server_tests.c

@@ -442,6 +442,93 @@ static void s_after_prepare_upload_part_finish(struct aws_s3_request *request) {
     }
 }
 
+/**
+ * This test is built for
+ * 1. The retry happens before the upload has finished.
+ */
+TEST_CASE(multipart_upload_checksum_with_retry_before_finish_mock_server) {
+    (void)ctx;
+    struct aws_s3_tester tester;
+    ASSERT_SUCCESS(aws_s3_tester_init(allocator, &tester));
+    struct aws_s3_tester_client_options client_options = {
+        .part_size = MB_TO_BYTES(5),
+        .tls_usage = AWS_S3_TLS_DISABLED,
+    };
+
+    struct aws_s3_client *client = NULL;
+    ASSERT_SUCCESS(aws_s3_tester_client_new(&tester, &client_options, &client));
+    struct aws_s3_client_vtable *patched_client_vtable = aws_s3_tester_patch_client_vtable(&tester, client, NULL);
+    patched_client_vtable->after_prepare_upload_part_finish = s_after_prepare_upload_part_finish;
+
+    struct aws_byte_cursor object_path = aws_byte_cursor_from_c_str("/throttle_before_finish");
+    {
+        /* 1. Trailer checksum */
+        struct aws_s3_tester_meta_request_options put_options = {
+            .allocator = allocator,
+            .meta_request_type = AWS_S3_META_REQUEST_TYPE_PUT_OBJECT,
+            .client = client,
+            .checksum_algorithm = AWS_SCA_CRC32,
+            .validate_get_response_checksum = false,
+            .put_options =
+                {
+                    .object_size_mb = 10,
+                    .object_path_override = object_path,
+                    .sleep_before_read_secs = 1,
+                },
+            .mock_server = true,
+        };
+
+        struct aws_s3_meta_request_test_results meta_request_test_results;
+        aws_s3_meta_request_test_results_init(&meta_request_test_results, allocator);
+
+        ASSERT_SUCCESS(aws_s3_tester_send_meta_request_with_options(&tester, &put_options, &meta_request_test_results));
+
+        ASSERT_INT_EQUALS(meta_request_test_results.upload_review.part_count, 2);
+        /* Note: the data we currently generate is always the same,
+         * so make sure that retry does not mangle the data by checking the checksum value */
+        ASSERT_STR_EQUALS(
+            "7/xUXw==", aws_string_c_str(meta_request_test_results.upload_review.part_checksums_array[0]));
+        ASSERT_STR_EQUALS(
+            "PCOjcw==", aws_string_c_str(meta_request_test_results.upload_review.part_checksums_array[1]));
+        aws_s3_meta_request_test_results_clean_up(&meta_request_test_results);
+    }
+    {
+        /* 2. header checksum */
+        struct aws_s3_tester_meta_request_options put_options = {
+            .allocator = allocator,
+            .meta_request_type = AWS_S3_META_REQUEST_TYPE_PUT_OBJECT,
+            .client = client,
+            .checksum_algorithm = AWS_SCA_CRC32,
+            .checksum_via_header = true,
+            .validate_get_response_checksum = false,
+            .put_options =
+                {
+                    .object_size_mb = 10,
+                    .object_path_override = object_path,
+                },
+            .mock_server = true,
+        };
+
+        struct aws_s3_meta_request_test_results meta_request_test_results;
+        aws_s3_meta_request_test_results_init(&meta_request_test_results, allocator);
+
+        ASSERT_SUCCESS(aws_s3_tester_send_meta_request_with_options(&tester, &put_options, &meta_request_test_results));
+
+        ASSERT_INT_EQUALS(meta_request_test_results.upload_review.part_count, 2);
+        /* Note: the data we currently generate is always the same,
+         * so make sure that retry does not mangle the data by checking the checksum value */
+        ASSERT_STR_EQUALS(
+            "7/xUXw==", aws_string_c_str(meta_request_test_results.upload_review.part_checksums_array[0]));
+        ASSERT_STR_EQUALS(
+            "PCOjcw==", aws_string_c_str(meta_request_test_results.upload_review.part_checksums_array[1]));
+        aws_s3_meta_request_test_results_clean_up(&meta_request_test_results);
+    }
+    aws_s3_client_release(client);
+    aws_s3_tester_clean_up(&tester);
+
+    return AWS_OP_SUCCESS;
+}
+
 /**
  * This test is built for
  * 1. We had a memory leak when the retry was triggered and the checksum was calculated.

tests/s3_tester.c

@@ -1681,6 +1681,7 @@ int aws_s3_tester_send_meta_request_with_options(
                         .autogen_length = upload_size_bytes,
                         .eof_requires_extra_read = options->put_options.eof_requires_extra_read,
                         .max_bytes_per_read = options->put_options.max_bytes_per_read,
+                        .sleep_before_read_secs = options->put_options.sleep_before_read_secs,
                     },
             };
             if (options->put_options.invalid_input_stream) {

tests/s3_tester.h

@@ -229,6 +229,7 @@ struct aws_s3_tester_meta_request_options {
         struct aws_byte_cursor content_encoding;
         enum aws_s3_tester_full_object_checksum full_object_checksum;
         struct aws_byte_cursor if_none_match_header;
+        size_t sleep_before_read_secs;
     } put_options;
 
     enum aws_s3_tester_sse_type sse_type;