diff --git a/include/aws/s3/private/s3_checksum_context.h b/include/aws/s3/private/s3_checksum_context.h
new file mode 100644
index 000000000..ce86aa977
--- /dev/null
+++ b/include/aws/s3/private/s3_checksum_context.h
@@ -0,0 +1,148 @@
+#ifndef AWS_S3_CHECKSUM_CONTEXT_H
+#define AWS_S3_CHECKSUM_CONTEXT_H
+
+/**
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include "aws/s3/s3_client.h"
+#include <aws/common/byte_buf.h>
+#include <aws/common/ref_count.h>
+
+struct aws_s3_meta_request_checksum_config_storage;
+
+AWS_EXTERN_C_BEGIN
+
+/**
+ * Upload request checksum context that encapsulates all checksum-related state and behavior
+ * for individual upload part requests. This replaces the complex tri-state buffer logic
+ * with a cleaner approach. Uses reference counting for lifetime management since context
+ * is transferred between functions.
+ */
+struct aws_s3_upload_request_checksum_context {
+    struct aws_allocator *allocator;
+    struct aws_ref_count ref_count;
+
+    /* Configuration */
+    enum aws_s3_checksum_algorithm algorithm;
+    enum aws_s3_checksum_location location;
+
+    struct aws_byte_buf base64_checksum;
+    /* The checksum already be calculated or not. */
+    bool checksum_calculated;
+
+    /* Validation */
+    size_t encoded_checksum_size;
+};
+
+/**
+ * Create a new upload request checksum context from configuration and buffer parameters.
+ * This function encapsulates all the complex logic for determining buffer state.
+ * Returns with reference count of 1.
+ *
+ * @param allocator Memory allocator
+ * @param checksum_config Meta request level checksum configuration (can be NULL)
+ * @return New checksum context or NULL on error
+ */
+AWS_S3_API
+struct aws_s3_upload_request_checksum_context *aws_s3_upload_request_checksum_context_new(
+    struct aws_allocator *allocator,
+    const struct aws_s3_meta_request_checksum_config_storage *checksum_config);
+
+/**
+ * Create a new upload request checksum context with an existing base64 encoded checksum value.
+ * This is useful when resuming uploads or when the checksum is pre-calculated.
+ * Returns with reference count of 1.
+ *
+ * @param allocator Memory allocator
+ * @param checksum_config Meta request level checksum configuration (can be NULL)
+ * @param existing_base64_checksum Pre-calculated checksum value as a byte cursor
+ * @return New checksum context or NULL on error (e.g., if checksum size doesn't match algorithm)
+ */
+AWS_S3_API
+struct aws_s3_upload_request_checksum_context *aws_s3_upload_request_checksum_context_new_with_existing_base64_checksum(
+    struct aws_allocator *allocator,
+    const struct aws_s3_meta_request_checksum_config_storage *checksum_config,
+    struct aws_byte_cursor existing_base64_checksum);
+
+/**
+ * Acquire a reference to the upload request checksum context.
+ * Use this when transferring ownership to another function/structure.
+ *
+ * @param context The checksum context to acquire
+ * @return The same context pointer (for convenience)
+ */
+AWS_S3_API
+struct aws_s3_upload_request_checksum_context *aws_s3_upload_request_checksum_context_acquire(
+    struct aws_s3_upload_request_checksum_context *context);
+
+/**
+ * Release a reference to the upload request checksum context.
+ * When the reference count reaches zero, the context will be destroyed.
+ * Always returns NULL.
+ *
+ * @param context The checksum context to release
+ */
+AWS_S3_API
+struct aws_s3_upload_request_checksum_context *aws_s3_upload_request_checksum_context_release(
+    struct aws_s3_upload_request_checksum_context *context);
+
+/**
+ * Check if checksum calculation is needed based on context state.
+ * Returns true if the context has a valid algorithm and the checksum has not been calculated yet.
+ *
+ * @param context The checksum context to check
+ * @return true if checksum calculation is needed, false otherwise
+ */
+AWS_S3_API
+bool aws_s3_upload_request_checksum_context_should_calculate(
+    const struct aws_s3_upload_request_checksum_context *context);
+
+/**
+ * Check if checksum should be added to HTTP headers.
+ * Returns true if the context has a valid algorithm and the location is set to header.
+ *
+ * @param context The checksum context to check
+ * @return true if checksum should be added to headers, false otherwise
+ */
+AWS_S3_API
+bool aws_s3_upload_request_checksum_context_should_add_header(
+    const struct aws_s3_upload_request_checksum_context *context);
+
+/**
+ * Check if checksum should be added as trailer (aws-chunked encoding).
+ * Returns true if the context has a valid algorithm and the location is set to trailer.
+ *
+ * @param context The checksum context to check
+ * @return true if checksum should be added as trailer, false otherwise
+ */
+AWS_S3_API
+bool aws_s3_upload_request_checksum_context_should_add_trailer(
+    const struct aws_s3_upload_request_checksum_context *context);
+
+/**
+ * Get the checksum buffer to use for output.
+ * Returns the internal buffer for storing the calculated checksum.
+ *
+ * @param context The checksum context
+ * @return Pointer to the checksum buffer, or NULL if context is invalid
+ */
+AWS_S3_API
+struct aws_byte_buf *aws_s3_upload_request_checksum_context_get_output_buffer(
+    struct aws_s3_upload_request_checksum_context *context);
+
+/**
+ * Get a cursor to the current base64 encoded checksum value (for use in headers/trailers).
+ * Returns an empty cursor if the checksum has not been calculated yet.
+ *
+ * @param context The checksum context
+ * @return Byte cursor to the calculated checksum, or empty cursor if not available
+ */
+AWS_S3_API
+struct aws_byte_cursor aws_s3_upload_request_checksum_context_get_checksum_cursor(
+    const struct aws_s3_upload_request_checksum_context *context);
+
+AWS_EXTERN_C_END
+
+#endif /* AWS_S3_CHECKSUM_CONTEXT_H */
diff --git a/include/aws/s3/private/s3_checksums.h b/include/aws/s3/private/s3_checksums.h
index a0193a626..6d922c84e 100644
--- a/include/aws/s3/private/s3_checksums.h
+++ b/include/aws/s3/private/s3_checksums.h
@@ -10,6 +10,7 @@
  * aws-c-sdkutil. */
 
 struct aws_s3_checksum;
+struct aws_s3_upload_request_checksum_context;
 
 /* List to check the checksum algorithm to use based on the priority. */
 static const enum aws_s3_checksum_algorithm s_checksum_algo_priority_list[] = {
@@ -40,7 +41,7 @@ struct aws_s3_checksum {
     } impl;
 };
 
-struct checksum_config_storage {
+struct aws_s3_meta_request_checksum_config_storage {
     struct aws_allocator *allocator;
     struct aws_byte_buf full_object_checksum;
     bool has_full_object_checksum;
@@ -83,25 +84,46 @@ struct aws_input_stream *aws_checksum_stream_new(
 
 /**
  * TODO: properly support chunked encoding.
+ * Creates a chunked encoding stream that wraps an existing stream and adds checksum trailers.
  *
- * A stream that takes in a stream, encodes it to aws_chunked. Computes a running checksum as it is read and add the
- * checksum as trailer at the end of the stream. All of the added bytes will be counted to the length of the stream.
- * Note: seek this stream will immediately fail, as it would prevent an accurate calculation of the
- * checksum.
+ * This function creates a stream that:
+ * 1. Encodes the input stream wraps the existing_stream with aws-chunked encoded.
+ * 2. Calculates a checksum of the stream content (if not already calculated)
+ * 3. Appends the checksum as a trailer at the end of the aws-chunked stream
  *
- * @param allocator
- * @param existing_stream   The data to be chunkified prepended by information on the stream length followed by a final
- *                          chunk and a trailing chunk containing a checksum of the existing stream. Destroying the
- *                          chunk stream will destroy the existing stream.
- * @param checksum_output   Optional argument, if provided the buffer will be initialized to the appropriate size and
- *                          filled with the checksum result when calculated. Callers responsibility to cleanup.
+ * Note: This stream does not support seeking operations, as seeking would prevent
+ * accurate checksum calculation and corrupt the chunked encoding format.
+ *
+ * @param allocator         Memory allocator to use for stream creation and internal buffers
+ * @param existing_stream   The input stream to be chunked and checksummed. This stream
+ *                          will be acquired by the chunk stream and released when the
+ *                          chunk stream is destroyed. Must not be NULL.
+ * @param checksum_context  Context containing checksum configuration and state. Must not be NULL.
+ *                          The context contains:
+ *                          - algorithm: The checksum algorithm to use (CRC32, CRC32C, etc.)
+ *                          - base64_checksum: Buffer for the calculated checksum result
+ *                          - checksum_calculated: Whether checksum is pre-calculated or needs calculation
+ *                          - encoded_checksum_size: Expected size of the base64-encoded checksum
+ *
+ *                          If checksum_calculated is false, the stream will wrap existing_stream
+ *                          with a checksum stream to calculate the checksum during reading.
+ *                          If checksum_calculated is true, the existing checksum value will be used.
+ *
+ * @return A new input stream that provides chunked encoding with checksum trailers,
+ *         or NULL if creation fails. The returned stream must be released with
+ *         aws_input_stream_release() when no longer needed.
+ *
+ * @note The total length of the returned stream includes:
+ *       - Chunk size header (hex representation + \r\n)
+ *       - Original stream content
+ *       - Final chunk marker (0\r\n or \r\n0\r\n)
+ *       - Checksum trailer (header name + : + base64 checksum + \r\n\r\n)
  */
 AWS_S3_API
 struct aws_input_stream *aws_chunk_stream_new(
     struct aws_allocator *allocator,
     struct aws_input_stream *existing_stream,
-    enum aws_s3_checksum_algorithm algorithm,
-    struct aws_byte_buf *checksum_output);
+    struct aws_s3_upload_request_checksum_context *context);
 
 /**
  * Get the size of the checksum output corresponding to the aws_s3_checksum_algorithm enum value.
@@ -166,14 +188,15 @@ AWS_S3_API
 int aws_checksum_finalize(struct aws_s3_checksum *checksum, struct aws_byte_buf *output);
 
 AWS_S3_API
-int aws_checksum_config_storage_init(
+int aws_s3_meta_request_checksum_config_storage_init(
     struct aws_allocator *allocator,
-    struct checksum_config_storage *internal_config,
+    struct aws_s3_meta_request_checksum_config_storage *internal_config,
     const struct aws_s3_checksum_config *config,
     const struct aws_http_message *message,
     const void *log_id);
 
 AWS_S3_API
-void aws_checksum_config_storage_cleanup(struct checksum_config_storage *internal_config);
+void aws_s3_meta_request_checksum_config_storage_cleanup(
+    struct aws_s3_meta_request_checksum_config_storage *internal_config);
 
 #endif /* AWS_S3_CHECKSUMS_H */
diff --git a/include/aws/s3/private/s3_client_impl.h b/include/aws/s3/private/s3_client_impl.h
index 1d3ecaa8a..c3520a7c8 100644
--- a/include/aws/s3/private/s3_client_impl.h
+++ b/include/aws/s3/private/s3_client_impl.h
@@ -174,6 +174,8 @@ struct aws_s3_client_vtable {
     struct aws_http_stream *(*http_connection_make_request)(
         struct aws_http_connection *client_connection,
         const struct aws_http_make_request_options *options);
+
+    void (*after_prepare_upload_part_finish)(struct aws_s3_request *request);
 };
 
 struct aws_s3_upload_part_timeout_stats {
diff --git a/include/aws/s3/private/s3_meta_request_impl.h b/include/aws/s3/private/s3_meta_request_impl.h
index 7db29a10a..3de1d4cac 100644
--- a/include/aws/s3/private/s3_meta_request_impl.h
+++ b/include/aws/s3/private/s3_meta_request_impl.h
@@ -25,7 +25,7 @@ struct aws_s3_request;
 struct aws_http_headers;
 struct aws_http_make_request_options;
 struct aws_retry_strategy;
-
+struct aws_s3_upload_request_checksum_context;
 enum aws_s3_meta_request_state {
     AWS_S3_META_REQUEST_STATE_ACTIVE,
     AWS_S3_META_REQUEST_STATE_FINISHED,
@@ -276,7 +276,7 @@ struct aws_s3_meta_request {
     const bool should_compute_content_md5;
 
     /* deep copy of the checksum config. */
-    struct checksum_config_storage checksum_config;
+    struct aws_s3_meta_request_checksum_config_storage checksum_config;
 
     /* checksum found in either a default get request, or in the initial head request of a multipart get */
     struct aws_byte_buf meta_request_level_response_header_checksum;
@@ -294,7 +294,7 @@ struct aws_s3_meta_request {
 struct aws_s3_mpu_part_info {
     uint64_t size;
     struct aws_string *etag;
-    struct aws_byte_buf checksum_base64;
+    struct aws_s3_upload_request_checksum_context *checksum_context;
     bool was_previously_uploaded;
 };
 
diff --git a/include/aws/s3/private/s3_request_messages.h b/include/aws/s3/private/s3_request_messages.h
index c477484aa..02a61c38f 100644
--- a/include/aws/s3/private/s3_request_messages.h
+++ b/include/aws/s3/private/s3_request_messages.h
@@ -17,7 +17,8 @@ struct aws_byte_buf;
 struct aws_byte_cursor;
 struct aws_string;
 struct aws_array_list;
-struct checksum_config_storage;
+struct aws_s3_meta_request_checksum_config_storage;
+struct aws_s3_upload_request_checksum_context;
 
 AWS_EXTERN_C_BEGIN
 
@@ -52,8 +53,7 @@ struct aws_input_stream *aws_s3_message_util_assign_body(
     struct aws_allocator *allocator,
     struct aws_byte_buf *byte_buf,
     struct aws_http_message *out_message,
-    const struct checksum_config_storage *checksum_config,
-    struct aws_byte_buf *out_checksum);
+    struct aws_s3_upload_request_checksum_context *checksum_context);
 
 /* Create an HTTP request for an S3 Ranged Get Object Request, using the given request as a basis */
 AWS_S3_API
@@ -76,7 +76,7 @@ AWS_S3_API
 struct aws_http_message *aws_s3_create_multipart_upload_message_new(
     struct aws_allocator *allocator,
     struct aws_http_message *base_message,
-    const struct checksum_config_storage *checksum_config);
+    const struct aws_s3_meta_request_checksum_config_storage *checksum_config);
 
 /* Create an HTTP request for an S3 Put Object request, using the original request as a basis.  Creates and assigns a
  * body stream using the passed in buffer.  If multipart is not needed, part number and upload_id can be 0 and NULL,
@@ -89,8 +89,7 @@ struct aws_http_message *aws_s3_upload_part_message_new(
     uint32_t part_number,
     const struct aws_string *upload_id,
     bool should_compute_content_md5,
-    const struct checksum_config_storage *checksum_config,
-    struct aws_byte_buf *encoded_checksum_output);
+    struct aws_s3_upload_request_checksum_context *checksum_context);
 
 /* Create an HTTP request for an S3 UploadPartCopy request, using the original request as a basis.
  * If multipart is not needed, part number and upload_id can be 0 and NULL,
@@ -116,7 +115,7 @@ struct aws_http_message *aws_s3_complete_multipart_message_new(
     struct aws_byte_buf *body_buffer,
     const struct aws_string *upload_id,
     const struct aws_array_list *parts,
-    const struct checksum_config_storage *checksum_config);
+    const struct aws_s3_meta_request_checksum_config_storage *checksum_config);
 
 AWS_S3_API
 struct aws_http_message *aws_s3_abort_multipart_upload_message_new(
diff --git a/source/s3_auto_ranged_put.c b/source/s3_auto_ranged_put.c
index 277d9bc4e..a6fd672cd 100644
--- a/source/s3_auto_ranged_put.c
+++ b/source/s3_auto_ranged_put.c
@@ -4,6 +4,7 @@
  */
 
 #include "aws/s3/private/s3_auto_ranged_put.h"
+#include "aws/s3/private/s3_checksum_context.h"
 #include "aws/s3/private/s3_checksums.h"
 #include "aws/s3/private/s3_list_parts.h"
 #include "aws/s3/private/s3_request_messages.h"
@@ -153,9 +154,17 @@ static int s_process_part_info_synced(const struct aws_s3_part_info *info, void
     }
 
     if ((checksum_cur != NULL) && (checksum_cur->len > 0)) {
-        aws_byte_buf_init_copy_from_cursor(&part->checksum_base64, auto_ranged_put->base.allocator, *checksum_cur);
+        /* Create checksum context with pre-calculated checksum */
+        part->checksum_context = aws_s3_upload_request_checksum_context_new_with_existing_base64_checksum(
+            auto_ranged_put->base.allocator, &auto_ranged_put->base.checksum_config, *checksum_cur);
+    } else {
+        part->checksum_context = aws_s3_upload_request_checksum_context_new(
+            auto_ranged_put->base.allocator, &auto_ranged_put->base.checksum_config);
+    }
+    if (part->checksum_context == NULL) {
+        aws_mem_release(meta_request->allocator, part);
+        return AWS_OP_ERR;
     }
-
     /* Parts might be out of order or have gaps in them.
      * Resize array-list to be long enough to hold this part,
      * filling any intermediate slots with NULL. */
@@ -406,7 +415,7 @@ static void s_s3_meta_request_auto_ranged_put_destroy(struct aws_s3_meta_request
         struct aws_s3_mpu_part_info *part;
         aws_array_list_get_at(&auto_ranged_put->synced_data.part_list, &part, part_index);
         if (part != NULL) {
-            aws_byte_buf_clean_up(&part->checksum_base64);
+            aws_s3_upload_request_checksum_context_release(part->checksum_context);
             aws_string_destroy(part->etag);
             aws_mem_release(auto_ranged_put->base.allocator, part);
         }
@@ -1067,6 +1076,8 @@ static void s_s3_prepare_upload_part_on_read_done(void *user_data) {
         /* Add part to array-list */
         struct aws_s3_mpu_part_info *part =
             aws_mem_calloc(meta_request->allocator, 1, sizeof(struct aws_s3_mpu_part_info));
+        part->checksum_context =
+            aws_s3_upload_request_checksum_context_new(meta_request->allocator, &meta_request->checksum_config);
         part->size = request->request_body.len;
         aws_array_list_set_at(&auto_ranged_put->synced_data.part_list, &part, request->part_number - 1);
     }
@@ -1087,13 +1098,14 @@ static void s_s3_prepare_upload_part_on_read_done(void *user_data) {
                 (void *)meta_request);
             goto on_done;
         }
+        struct aws_s3_upload_request_checksum_context *context = previously_uploaded_info->checksum_context;
         /* if previously uploaded part had a checksum, compare it to what we just skipped */
-        if (previously_uploaded_info->checksum_base64.len > 0 &&
+        if (context != NULL && context->checksum_calculated == true &&
             s_verify_part_matches_checksum(
                 meta_request->allocator,
                 aws_byte_cursor_from_buf(&request->request_body),
                 meta_request->checksum_config.checksum_algorithm,
-                aws_byte_cursor_from_buf(&previously_uploaded_info->checksum_base64))) {
+                aws_byte_cursor_from_buf(&context->base64_checksum))) {
             error_code = aws_last_error_or_unknown();
             goto on_done;
         }
@@ -1115,6 +1127,7 @@ static void s_s3_prepare_upload_part_on_read_done(void *user_data) {
 static void s_s3_prepare_upload_part_finish(struct aws_s3_prepare_upload_part_job *part_prep, int error_code) {
     struct aws_s3_request *request = part_prep->request;
     struct aws_s3_meta_request *meta_request = request->meta_request;
+    struct aws_s3_client *client = meta_request->client;
     struct aws_s3_auto_ranged_put *auto_ranged_put = meta_request->impl;
 
     if (error_code != AWS_ERROR_SUCCESS) {
@@ -1122,7 +1135,7 @@ static void s_s3_prepare_upload_part_finish(struct aws_s3_prepare_upload_part_jo
         goto on_done;
     }
 
-    struct aws_byte_buf *checksum_buf = NULL;
+    struct aws_s3_upload_request_checksum_context *checksum_context = NULL;
     if (request->is_noop) {
         AWS_LOGF_DEBUG(
             AWS_LS_S3_META_REQUEST,
@@ -1139,10 +1152,14 @@ static void s_s3_prepare_upload_part_finish(struct aws_s3_prepare_upload_part_jo
             aws_s3_meta_request_lock_synced_data(meta_request);
             struct aws_s3_mpu_part_info *part = NULL;
             aws_array_list_get_at(&auto_ranged_put->synced_data.part_list, &part, request->part_number - 1);
-            AWS_ASSERT(part != NULL);
-            checksum_buf = &part->checksum_base64;
-            /* Clean up the buffer in case of it's initialized before and retry happens. */
-            aws_byte_buf_clean_up(checksum_buf);
+            AWS_ASSERT(part != NULL && part->checksum_context != NULL);
+            /* Use checksum context if available, otherwise NULL for new parts */
+            checksum_context = part->checksum_context;
+            /* If checksum already calculated, it means either the part being retried or the part resumed from list
+             * parts. Keep reusing the old checksum in case of the request body in memory mangled */
+            AWS_ASSERT(
+                !checksum_context->checksum_calculated || request->num_times_prepared > 0 ||
+                auto_ranged_put->resume_token != NULL);
             aws_s3_meta_request_unlock_synced_data(meta_request);
         }
         /* END CRITICAL SECTION */
@@ -1162,8 +1179,7 @@ static void s_s3_prepare_upload_part_finish(struct aws_s3_prepare_upload_part_jo
         request->part_number,
         auto_ranged_put->upload_id,
         meta_request->should_compute_content_md5,
-        &meta_request->checksum_config,
-        checksum_buf);
+        checksum_context);
     if (message == NULL) {
         aws_future_http_message_set_error(part_prep->on_complete, aws_last_error());
         goto on_done;
@@ -1173,6 +1189,10 @@ static void s_s3_prepare_upload_part_finish(struct aws_s3_prepare_upload_part_jo
     aws_future_http_message_set_result_by_move(part_prep->on_complete, &message);
 
 on_done:
+    if (client->vtable->after_prepare_upload_part_finish) {
+        /* TEST ONLY, allow test to stub here. */
+        client->vtable->after_prepare_upload_part_finish(request);
+    }
     AWS_FATAL_ASSERT(aws_future_http_message_is_done(part_prep->on_complete));
     aws_future_bool_release(part_prep->asyncstep_read_part);
     aws_future_http_message_release(part_prep->on_complete);
@@ -1209,7 +1229,12 @@ static int s_s3_review_multipart_upload(struct aws_s3_request *request) {
 
             struct aws_s3_upload_part_review *part_review = &review.part_array[part_index];
             part_review->size = part->size;
-            part_review->checksum = aws_byte_cursor_from_buf(&part->checksum_base64);
+            if (part->checksum_context != NULL) {
+                part_review->checksum =
+                    aws_s3_upload_request_checksum_context_get_checksum_cursor(part->checksum_context);
+            } else {
+                part_review->checksum = (struct aws_byte_cursor){.ptr = NULL, .len = 0};
+            }
         }
     }
 
diff --git a/source/s3_checksum_context.c b/source/s3_checksum_context.c
new file mode 100644
index 000000000..189a2b128
--- /dev/null
+++ b/source/s3_checksum_context.c
@@ -0,0 +1,147 @@
+/**
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+
+#include "aws/s3/private/s3_checksum_context.h"
+#include "aws/s3/private/s3_checksums.h"
+#include <aws/common/encoding.h>
+#include <aws/common/logging.h>
+
+static void s_aws_s3_upload_request_checksum_context_destroy(void *context) {
+    struct aws_s3_upload_request_checksum_context *checksum_context = context;
+    aws_byte_buf_clean_up(&checksum_context->base64_checksum);
+    aws_mem_release(checksum_context->allocator, checksum_context);
+}
+
+static struct aws_s3_upload_request_checksum_context *s_s3_upload_request_checksum_context_new_base(
+    struct aws_allocator *allocator,
+    const struct aws_s3_meta_request_checksum_config_storage *checksum_config) {
+    AWS_PRECONDITION(allocator);
+
+    struct aws_s3_upload_request_checksum_context *context =
+        aws_mem_calloc(allocator, 1, sizeof(struct aws_s3_upload_request_checksum_context));
+
+    aws_ref_count_init(&context->ref_count, context, s_aws_s3_upload_request_checksum_context_destroy);
+    context->allocator = allocator;
+    /* Handle case where no checksum config is provided */
+    if (!checksum_config || checksum_config->checksum_algorithm == AWS_SCA_NONE) {
+        context->algorithm = AWS_SCA_NONE;
+        context->encoded_checksum_size = 0;
+        return context;
+    }
+
+    /* Extract configuration */
+    context->algorithm = checksum_config->checksum_algorithm;
+    context->location = checksum_config->location;
+    context->encoded_checksum_size = aws_get_digest_size_from_checksum_algorithm(context->algorithm);
+
+    /* Convert to base64 encoded size */
+    size_t encoded_size = 0;
+    if (aws_base64_compute_encoded_len(context->encoded_checksum_size, &encoded_size)) {
+        AWS_LOGF_ERROR(AWS_LS_S3_GENERAL, "Failed to compute base64 encoded length for checksum");
+        aws_s3_upload_request_checksum_context_release(context);
+        return NULL;
+    }
+    context->encoded_checksum_size = encoded_size;
+    return context;
+}
+
+struct aws_s3_upload_request_checksum_context *aws_s3_upload_request_checksum_context_new(
+    struct aws_allocator *allocator,
+    const struct aws_s3_meta_request_checksum_config_storage *checksum_config) {
+    struct aws_s3_upload_request_checksum_context *context =
+        s_s3_upload_request_checksum_context_new_base(allocator, checksum_config);
+    if (context && context->encoded_checksum_size > 0) {
+        /* Initial the buffer for checksum */
+        aws_byte_buf_init(&context->base64_checksum, allocator, context->encoded_checksum_size);
+    }
+    return context;
+}
+
+struct aws_s3_upload_request_checksum_context *aws_s3_upload_request_checksum_context_new_with_existing_base64_checksum(
+    struct aws_allocator *allocator,
+    const struct aws_s3_meta_request_checksum_config_storage *checksum_config,
+    struct aws_byte_cursor existing_base64_checksum) {
+    struct aws_s3_upload_request_checksum_context *context =
+        s_s3_upload_request_checksum_context_new_base(allocator, checksum_config);
+    if (context) {
+        /* Initial the buffer for checksum from the exist checksum */
+        if (context->encoded_checksum_size != existing_base64_checksum.len) {
+            struct aws_byte_cursor algo_name = aws_get_checksum_algorithm_name(context->algorithm);
+            AWS_LOGF_ERROR(
+                AWS_LS_S3_GENERAL,
+                "Encoded checksum size mismatch during creating the context for algorithm " PRInSTR
+                ": expected %zu bytes, got %zu bytes",
+                AWS_BYTE_CURSOR_PRI(algo_name),
+                context->encoded_checksum_size,
+                existing_base64_checksum.len);
+            aws_s3_upload_request_checksum_context_release(context);
+            return NULL;
+        }
+        aws_byte_buf_init_copy_from_cursor(&context->base64_checksum, allocator, existing_base64_checksum);
+        context->checksum_calculated = true;
+    }
+    return context;
+}
+
+struct aws_s3_upload_request_checksum_context *aws_s3_upload_request_checksum_context_acquire(
+    struct aws_s3_upload_request_checksum_context *context) {
+    if (context) {
+        aws_ref_count_acquire(&context->ref_count);
+    }
+    return context;
+}
+
+struct aws_s3_upload_request_checksum_context *aws_s3_upload_request_checksum_context_release(
+    struct aws_s3_upload_request_checksum_context *context) {
+    if (context) {
+        aws_ref_count_release(&context->ref_count);
+    }
+    return NULL;
+}
+
+bool aws_s3_upload_request_checksum_context_should_calculate(
+    const struct aws_s3_upload_request_checksum_context *context) {
+    if (!context || context->algorithm == AWS_SCA_NONE) {
+        return false;
+    }
+
+    /* If not previous calculated */
+    return !context->checksum_calculated;
+}
+
+bool aws_s3_upload_request_checksum_context_should_add_header(
+    const struct aws_s3_upload_request_checksum_context *context) {
+    if (!context) {
+        return false;
+    }
+
+    return context->location == AWS_SCL_HEADER && context->algorithm != AWS_SCA_NONE;
+}
+
+bool aws_s3_upload_request_checksum_context_should_add_trailer(
+    const struct aws_s3_upload_request_checksum_context *context) {
+    if (!context) {
+        return false;
+    }
+
+    return context->location == AWS_SCL_TRAILER && context->algorithm != AWS_SCA_NONE;
+}
+
+struct aws_byte_buf *aws_s3_upload_request_checksum_context_get_output_buffer(
+    struct aws_s3_upload_request_checksum_context *context) {
+    if (!context) {
+        return NULL;
+    }
+    return &context->base64_checksum;
+}
+
+struct aws_byte_cursor aws_s3_upload_request_checksum_context_get_checksum_cursor(
+    const struct aws_s3_upload_request_checksum_context *context) {
+    struct aws_byte_cursor checksum_cursor = {0};
+    if (!context || !context->checksum_calculated) {
+        return checksum_cursor;
+    }
+    return aws_byte_cursor_from_buf(&context->base64_checksum);
+}
diff --git a/source/s3_checksum_stream.c b/source/s3_checksum_stream.c
index 2b21351af..4647073a8 100644
--- a/source/s3_checksum_stream.c
+++ b/source/s3_checksum_stream.c
@@ -122,9 +122,10 @@ struct aws_input_stream *aws_checksum_stream_new(
     enum aws_s3_checksum_algorithm algorithm,
     struct aws_byte_buf *checksum_output) {
     AWS_PRECONDITION(existing_stream);
+    AWS_PRECONDITION(checksum_output);
+    AWS_PRECONDITION(checksum_output->len == 0 && "Checksum output buffer is not empty");
 
     struct aws_checksum_stream *impl = aws_mem_calloc(allocator, 1, sizeof(struct aws_checksum_stream));
-
     impl->allocator = allocator;
     impl->base.vtable = &s_aws_input_checksum_stream_vtable;
 
diff --git a/source/s3_checksums.c b/source/s3_checksums.c
index d1d561ad8..69007b522 100644
--- a/source/s3_checksums.c
+++ b/source/s3_checksums.c
@@ -332,7 +332,7 @@ int aws_checksum_compute(
 }
 
 static int s_init_and_verify_checksum_config_from_headers(
-    struct checksum_config_storage *checksum_config,
+    struct aws_s3_meta_request_checksum_config_storage *checksum_config,
     const struct aws_http_message *message,
     const void *log_id) {
     /* Check if the checksum header was set from the message */
@@ -394,9 +394,9 @@ static int s_init_and_verify_checksum_config_from_headers(
     return AWS_OP_SUCCESS;
 }
 
-int aws_checksum_config_storage_init(
+int aws_s3_meta_request_checksum_config_storage_init(
     struct aws_allocator *allocator,
-    struct checksum_config_storage *internal_config,
+    struct aws_s3_meta_request_checksum_config_storage *internal_config,
     const struct aws_s3_checksum_config *config,
     const struct aws_http_message *message,
     const void *log_id) {
@@ -488,7 +488,8 @@ int aws_checksum_config_storage_init(
     return AWS_OP_SUCCESS;
 }
 
-void aws_checksum_config_storage_cleanup(struct checksum_config_storage *internal_config) {
+void aws_s3_meta_request_checksum_config_storage_cleanup(
+    struct aws_s3_meta_request_checksum_config_storage *internal_config) {
     if (internal_config->has_full_object_checksum) {
         aws_byte_buf_clean_up(&internal_config->full_object_checksum);
     }
diff --git a/source/s3_chunk_stream.c b/source/s3_chunk_stream.c
index 273cfc77b..0b00d532b 100644
--- a/source/s3_chunk_stream.c
+++ b/source/s3_chunk_stream.c
@@ -3,6 +3,7 @@
  * SPDX-License-Identifier: Apache-2.0.
  */
 
+#include "aws/s3/private/s3_checksum_context.h"
 #include "aws/s3/private/s3_checksums.h"
 #include <aws/common/encoding.h>
 #include <aws/common/string.h>
@@ -22,14 +23,12 @@ typedef int(set_stream_fn)(struct aws_chunk_stream *parent_stream);
 struct aws_chunk_stream {
     struct aws_input_stream base;
     struct aws_allocator *allocator;
-
     /* aws_input_stream_byte_cursor provides our actual functionality  */
     /* Pointing to the stream we read from */
     struct aws_input_stream *current_stream;
+    struct aws_input_stream *chunk_body_stream;
 
-    struct aws_input_stream *checksum_stream;
-    struct aws_byte_buf checksum_result;
-    struct aws_byte_buf *checksum_result_output;
+    struct aws_s3_upload_request_checksum_context *checksum_context;
     struct aws_byte_buf pre_chunk_buffer;
     struct aws_byte_buf post_chunk_buffer;
     struct aws_byte_cursor checksum_header_name;
@@ -62,17 +61,12 @@ static int s_set_post_chunk_stream(struct aws_chunk_stream *parent_stream) {
     }
     struct aws_byte_cursor post_trailer_cursor = aws_byte_cursor_from_string(s_post_trailer);
     struct aws_byte_cursor colon_cursor = aws_byte_cursor_from_string(s_colon);
+    /* After the checksum stream released, the checksum will be calculated. */
+    parent_stream->checksum_context->checksum_calculated = true;
+    struct aws_byte_cursor checksum_result_cursor =
+        aws_byte_cursor_from_buf(&parent_stream->checksum_context->base64_checksum);
+    AWS_ASSERT(parent_stream->checksum_context->encoded_checksum_size == checksum_result_cursor.len);
 
-    if (parent_stream->checksum_result.len == 0) {
-        AWS_LOGF_ERROR(AWS_LS_S3_META_REQUEST, "Failed to extract base64 encoded checksum of stream");
-        return aws_raise_error(AWS_ERROR_S3_CHECKSUM_CALCULATION_FAILED);
-    }
-    struct aws_byte_cursor checksum_result_cursor = aws_byte_cursor_from_buf(&parent_stream->checksum_result);
-    if (parent_stream->checksum_result_output &&
-        aws_byte_buf_init_copy_from_cursor(
-            parent_stream->checksum_result_output, parent_stream->allocator, checksum_result_cursor)) {
-        return AWS_OP_ERR;
-    }
     if (aws_byte_buf_init(
             &parent_stream->post_chunk_buffer,
             parent_stream->allocator,
@@ -92,16 +86,15 @@ static int s_set_post_chunk_stream(struct aws_chunk_stream *parent_stream) {
     parent_stream->set_current_stream_fn = s_set_null_stream;
     return AWS_OP_SUCCESS;
 error:
-    aws_byte_buf_clean_up(parent_stream->checksum_result_output);
     aws_byte_buf_clean_up(&parent_stream->post_chunk_buffer);
     return AWS_OP_ERR;
 }
 
 static int s_set_chunk_stream(struct aws_chunk_stream *parent_stream) {
     aws_input_stream_release(parent_stream->current_stream);
-    parent_stream->current_stream = parent_stream->checksum_stream;
+    parent_stream->current_stream = parent_stream->chunk_body_stream;
     aws_byte_buf_clean_up(&parent_stream->pre_chunk_buffer);
-    parent_stream->checksum_stream = NULL;
+    parent_stream->chunk_body_stream = NULL;
     parent_stream->set_current_stream_fn = s_set_post_chunk_stream;
     return AWS_OP_SUCCESS;
 }
@@ -169,12 +162,12 @@ static void s_aws_input_chunk_stream_destroy(struct aws_chunk_stream *impl) {
         if (impl->current_stream) {
             aws_input_stream_release(impl->current_stream);
         }
-        if (impl->checksum_stream) {
-            aws_input_stream_release(impl->checksum_stream);
+        if (impl->chunk_body_stream) {
+            aws_input_stream_release(impl->chunk_body_stream);
         }
         aws_byte_buf_clean_up(&impl->pre_chunk_buffer);
-        aws_byte_buf_clean_up(&impl->checksum_result);
         aws_byte_buf_clean_up(&impl->post_chunk_buffer);
+        aws_s3_upload_request_checksum_context_release(impl->checksum_context);
         aws_mem_release(impl->allocator, impl);
     }
 }
@@ -189,14 +182,26 @@ static struct aws_input_stream_vtable s_aws_input_chunk_stream_vtable = {
 struct aws_input_stream *aws_chunk_stream_new(
     struct aws_allocator *allocator,
     struct aws_input_stream *existing_stream,
-    enum aws_s3_checksum_algorithm algorithm,
-    struct aws_byte_buf *checksum_output) {
+    struct aws_s3_upload_request_checksum_context *checksum_context) {
+    AWS_PRECONDITION(allocator);
+    AWS_PRECONDITION(existing_stream);
+    AWS_PRECONDITION(checksum_context);
 
     struct aws_chunk_stream *impl = aws_mem_calloc(allocator, 1, sizeof(struct aws_chunk_stream));
 
     impl->allocator = allocator;
     impl->base.vtable = &s_aws_input_chunk_stream_vtable;
-    impl->checksum_result_output = checksum_output;
+
+    /* Extract algorithm and buffer from context */
+    enum aws_s3_checksum_algorithm algorithm = AWS_SCA_NONE;
+    struct aws_byte_buf *checksum_buffer = NULL;
+
+    impl->checksum_context = aws_s3_upload_request_checksum_context_acquire(checksum_context);
+
+    algorithm = checksum_context->algorithm;
+    checksum_buffer = &checksum_context->base64_checksum;
+    bool checksum_calculated = checksum_context->checksum_calculated;
+
     int64_t stream_length = 0;
     int64_t final_chunk_len = 0;
     if (aws_input_stream_get_length(existing_stream, &stream_length)) {
@@ -218,22 +223,16 @@ struct aws_input_stream *aws_chunk_stream_new(
     if (aws_byte_buf_append(&impl->pre_chunk_buffer, &pre_chunk_cursor)) {
         goto error;
     }
-
-    size_t checksum_len = aws_get_digest_size_from_checksum_algorithm(algorithm);
-
-    size_t encoded_checksum_len = 0;
-    if (aws_base64_compute_encoded_len(checksum_len, &encoded_checksum_len)) {
-        goto error;
-    }
-    if (aws_byte_buf_init(&impl->checksum_result, allocator, encoded_checksum_len)) {
-        goto error;
-    }
-
-    impl->checksum_stream = aws_checksum_stream_new(allocator, existing_stream, algorithm, &impl->checksum_result);
-    if (impl->checksum_stream == NULL) {
-        goto error;
+    if (!checksum_calculated) {
+        /* Wrap the existing stream with checksum stream to calculate the checksum when reading from it. */
+        impl->chunk_body_stream = aws_checksum_stream_new(allocator, existing_stream, algorithm, checksum_buffer);
+        if (impl->chunk_body_stream == NULL) {
+            goto error;
+        }
+    } else {
+        /* No need to calculate the checksum during read, use the existing stream directly. */
+        impl->chunk_body_stream = aws_input_stream_acquire(existing_stream);
     }
-
     int64_t prechunk_stream_len = 0;
     int64_t colon_len = s_colon->len;
     int64_t post_trailer_len = s_post_trailer->len;
@@ -248,9 +247,9 @@ struct aws_input_stream *aws_chunk_stream_new(
         }
         impl->set_current_stream_fn = s_set_chunk_stream;
     } else {
-        impl->current_stream = impl->checksum_stream;
+        impl->current_stream = impl->chunk_body_stream;
         final_chunk_len = s_empty_chunk->len;
-        impl->checksum_stream = NULL;
+        impl->chunk_body_stream = NULL;
         impl->set_current_stream_fn = s_set_post_chunk_stream;
     }
 
@@ -261,17 +260,16 @@ struct aws_input_stream *aws_chunk_stream_new(
     }
 
     impl->length = prechunk_stream_len + stream_length + final_chunk_len + impl->checksum_header_name.len + colon_len +
-                   encoded_checksum_len + post_trailer_len;
+                   checksum_context->encoded_checksum_size + post_trailer_len;
 
     AWS_ASSERT(impl->current_stream);
     aws_ref_count_init(&impl->base.ref_count, impl, (aws_simple_completion_callback *)s_aws_input_chunk_stream_destroy);
     return &impl->base;
 
 error:
-    aws_input_stream_release(impl->checksum_stream);
+    aws_input_stream_release(impl->chunk_body_stream);
     aws_input_stream_release(impl->current_stream);
     aws_byte_buf_clean_up(&impl->pre_chunk_buffer);
-    aws_byte_buf_clean_up(&impl->checksum_result);
     aws_mem_release(impl->allocator, impl);
     return NULL;
 }
diff --git a/source/s3_copy_object.c b/source/s3_copy_object.c
index acfc81f05..3e7176151 100644
--- a/source/s3_copy_object.c
+++ b/source/s3_copy_object.c
@@ -4,6 +4,7 @@
  */
 
 #include "aws/s3/private/s3_copy_object.h"
+#include "aws/s3/private/s3_checksum_context.h"
 #include "aws/s3/private/s3_request_messages.h"
 #include "aws/s3/private/s3_util.h"
 #include <aws/common/string.h>
@@ -124,7 +125,7 @@ static void s_s3_meta_request_copy_object_destroy(struct aws_s3_meta_request *me
         struct aws_s3_mpu_part_info *part = NULL;
         aws_array_list_get_at(&copy_object->synced_data.part_list, &part, part_index);
         aws_string_destroy(part->etag);
-        aws_byte_buf_clean_up(&part->checksum_base64);
+        aws_s3_upload_request_checksum_context_release(part->checksum_context);
         aws_mem_release(meta_request->allocator, part);
     }
 
diff --git a/source/s3_default_meta_request.c b/source/s3_default_meta_request.c
index 074a67425..66992745f 100644
--- a/source/s3_default_meta_request.c
+++ b/source/s3_default_meta_request.c
@@ -1,4 +1,5 @@
 #include "aws/s3/private/s3_default_meta_request.h"
+#include "aws/s3/private/s3_checksum_context.h"
 #include "aws/s3/private/s3_client_impl.h"
 #include "aws/s3/private/s3_meta_request_impl.h"
 #include "aws/s3/private/s3_request_messages.h"
@@ -352,12 +353,14 @@ static void s_s3_default_prepare_request_finish(
         meta_request_default->request_type == AWS_S3_REQUEST_TYPE_UPLOAD_PART || request->request_body.len > 0) {
         /* Only PUT Object and Upload part support trailing checksum, that needs the special encoding even if the body
          * has 0 length. */
-        aws_s3_message_util_assign_body(
-            meta_request->allocator,
-            &request->request_body,
-            message,
-            &meta_request->checksum_config,
-            NULL /* out_checksum */);
+        /* Create checksum context from config if needed */
+        struct aws_s3_upload_request_checksum_context *checksum_context =
+            aws_s3_upload_request_checksum_context_new(meta_request->allocator, &meta_request->checksum_config);
+
+        aws_s3_message_util_assign_body(meta_request->allocator, &request->request_body, message, checksum_context);
+
+        /* Release the context reference */
+        aws_s3_upload_request_checksum_context_release(checksum_context);
     }
     aws_s3_request_setup_send_data(request, message);
 
diff --git a/source/s3_meta_request.c b/source/s3_meta_request.c
index 0ef22c471..50d856661 100644
--- a/source/s3_meta_request.c
+++ b/source/s3_meta_request.c
@@ -228,7 +228,7 @@ int aws_s3_meta_request_init_base(
 
     *((size_t *)&meta_request->part_size) = part_size;
     *((bool *)&meta_request->should_compute_content_md5) = should_compute_content_md5;
-    if (aws_checksum_config_storage_init(
+    if (aws_s3_meta_request_checksum_config_storage_init(
             meta_request->allocator,
             &meta_request->checksum_config,
             options->checksum_config,
@@ -500,7 +500,7 @@ static void s_s3_meta_request_destroy(void *user_data) {
     AWS_LOGF_DEBUG(AWS_LS_S3_META_REQUEST, "id=%p Cleaning up meta request", (void *)meta_request);
 
     /* Clean up our initial http message */
-    aws_checksum_config_storage_cleanup(&meta_request->checksum_config);
+    aws_s3_meta_request_checksum_config_storage_cleanup(&meta_request->checksum_config);
     meta_request->request_body_async_stream = aws_async_input_stream_release(meta_request->request_body_async_stream);
     meta_request->initial_request_message = aws_http_message_release(meta_request->initial_request_message);
 
diff --git a/source/s3_request_messages.c b/source/s3_request_messages.c
index 7776a487d..3e5a35693 100644
--- a/source/s3_request_messages.c
+++ b/source/s3_request_messages.c
@@ -4,6 +4,7 @@
  */
 
 #include "aws/s3/private/s3_request_messages.h"
+#include "aws/s3/private/s3_checksum_context.h"
 #include "aws/s3/private/s3_meta_request_impl.h"
 #include "aws/s3/private/s3_util.h"
 #include <aws/cal/hash.h>
@@ -286,7 +287,7 @@ struct aws_http_message *aws_s3_ranged_get_object_message_new(
 struct aws_http_message *aws_s3_create_multipart_upload_message_new(
     struct aws_allocator *allocator,
     struct aws_http_message *base_message,
-    const struct checksum_config_storage *checksum_config) {
+    const struct aws_s3_meta_request_checksum_config_storage *checksum_config) {
     AWS_PRECONDITION(allocator);
 
     /* For multipart upload, some headers should ONLY be in the initial create-multipart request.
@@ -352,8 +353,10 @@ struct aws_http_message *aws_s3_create_multipart_upload_message_new(
     return NULL;
 }
 
-/* Create a new put object request from an existing put object request.  Currently just optionally adds part information
- * for a multipart upload. */
+/**
+ * Create a new put object request from an existing put object request.  Currently just optionally adds part information
+ * for a multipart upload.
+ **/
 struct aws_http_message *aws_s3_upload_part_message_new(
     struct aws_allocator *allocator,
     struct aws_http_message *base_message,
@@ -361,8 +364,7 @@ struct aws_http_message *aws_s3_upload_part_message_new(
     uint32_t part_number,
     const struct aws_string *upload_id,
     bool should_compute_content_md5,
-    const struct checksum_config_storage *checksum_config,
-    struct aws_byte_buf *encoded_checksum_output) {
+    struct aws_s3_upload_request_checksum_context *checksum_context) {
     AWS_PRECONDITION(allocator);
     AWS_PRECONDITION(base_message);
     AWS_PRECONDITION(part_number > 0);
@@ -383,12 +385,12 @@ struct aws_http_message *aws_s3_upload_part_message_new(
         goto error_clean_up;
     }
 
-    if (aws_s3_message_util_assign_body(allocator, buffer, message, checksum_config, encoded_checksum_output) == NULL) {
+    if (aws_s3_message_util_assign_body(allocator, buffer, message, checksum_context) == NULL) {
         goto error_clean_up;
     }
 
     if (should_compute_content_md5) {
-        if (!checksum_config || checksum_config->location == AWS_SCL_NONE) {
+        if (!checksum_context || checksum_context->location == AWS_SCL_NONE) {
             /* MD5 will be skipped if flexible checksum used */
             if (aws_s3_message_util_add_content_md5_header(allocator, buffer, message)) {
                 goto error_clean_up;
@@ -434,8 +436,7 @@ struct aws_http_message *aws_s3_upload_part_copy_message_new(
     if (buffer != NULL) {
         /* part copy does not have a ChecksumAlgorithm member, it will use the same algorithm as the create
          * multipart upload request specifies */
-        if (aws_s3_message_util_assign_body(
-                allocator, buffer, message, NULL /* checksum_config */, NULL /* out_checksum */) == NULL) {
+        if (aws_s3_message_util_assign_body(allocator, buffer, message, NULL /*checksum_context*/) == NULL) {
             goto error_clean_up;
         }
 
@@ -628,7 +629,7 @@ struct aws_http_message *aws_s3_complete_multipart_message_new(
     struct aws_byte_buf *body_buffer,
     const struct aws_string *upload_id,
     const struct aws_array_list *parts,
-    const struct checksum_config_storage *checksum_config) {
+    const struct aws_s3_meta_request_checksum_config_storage *checksum_config) {
     AWS_PRECONDITION(allocator);
     AWS_PRECONDITION(base_message);
     AWS_PRECONDITION(body_buffer);
@@ -744,7 +745,8 @@ struct aws_http_message *aws_s3_complete_multipart_message_new(
             }
 
             if (mpu_algorithm_checksum_name.len) {
-                struct aws_byte_cursor checksum = aws_byte_cursor_from_buf(&part->checksum_base64);
+                struct aws_byte_cursor checksum =
+                    aws_s3_upload_request_checksum_context_get_checksum_cursor(part->checksum_context);
 
                 if (aws_byte_buf_append_dynamic(body_buffer, &s_open_start_bracket)) {
                     goto error_clean_up;
@@ -779,8 +781,7 @@ struct aws_http_message *aws_s3_complete_multipart_message_new(
 
         AWS_LOGF_TRACE(
             AWS_LS_S3_GENERAL, "Payload for Complete MPU is:\n" PRInSTR "\n", AWS_BYTE_BUF_PRI(*body_buffer));
-        aws_s3_message_util_assign_body(
-            allocator, body_buffer, message, NULL /* checksum_config */, NULL /* out_checksum */);
+        aws_s3_message_util_assign_body(allocator, body_buffer, message, NULL);
     }
 
     return message;
@@ -835,36 +836,36 @@ struct aws_http_message *aws_s3_abort_multipart_upload_message_new(
 static int s_calculate_in_memory_checksum_helper(
     struct aws_allocator *allocator,
     struct aws_byte_cursor data,
-    const struct checksum_config_storage *checksum_config,
-    struct aws_byte_buf *out_checksum) {
-    AWS_ASSERT(checksum_config->checksum_algorithm != AWS_SCA_NONE);
-    AWS_ASSERT(out_checksum != NULL);
-    AWS_ZERO_STRUCT(*out_checksum);
-
-    int ret_code = AWS_OP_ERR;
-    size_t digest_size = aws_get_digest_size_from_checksum_algorithm(checksum_config->checksum_algorithm);
-    size_t encoded_checksum_len = 0;
-    if (aws_base64_compute_encoded_len(digest_size, &encoded_checksum_len)) {
-        return AWS_OP_ERR;
+    struct aws_s3_upload_request_checksum_context *checksum_context) {
+    AWS_ASSERT(checksum_context);
+    if (checksum_context->checksum_calculated) {
+        return AWS_OP_SUCCESS;
     }
 
-    aws_byte_buf_init(out_checksum, allocator, encoded_checksum_len);
-
+    int ret_code = AWS_OP_ERR;
+    /* Calculate checksum for output buffer only (no header/trailer) */
+    struct aws_byte_buf *output_buffer = aws_s3_upload_request_checksum_context_get_output_buffer(checksum_context);
     struct aws_byte_buf raw_checksum;
+    size_t digest_size = aws_get_digest_size_from_checksum_algorithm(checksum_context->algorithm);
     aws_byte_buf_init(&raw_checksum, allocator, digest_size);
 
-    if (aws_checksum_compute(allocator, checksum_config->checksum_algorithm, &data, &raw_checksum)) {
+    if (aws_checksum_compute(allocator, checksum_context->algorithm, &data, &raw_checksum)) {
+        aws_byte_buf_clean_up(&raw_checksum);
         goto done;
     }
+
     struct aws_byte_cursor raw_checksum_cursor = aws_byte_cursor_from_buf(&raw_checksum);
-    if (aws_base64_encode(&raw_checksum_cursor, out_checksum)) {
+    if (aws_base64_encode(&raw_checksum_cursor, output_buffer)) {
+        aws_byte_buf_clean_up(&raw_checksum);
         goto done;
     }
+    aws_byte_buf_clean_up(&raw_checksum);
+    checksum_context->checksum_calculated = true;
 
     ret_code = AWS_OP_SUCCESS;
 done:
     if (ret_code) {
-        aws_byte_buf_clean_up(out_checksum);
+        aws_byte_buf_clean_up(output_buffer);
     }
     aws_byte_buf_clean_up(&raw_checksum);
     return ret_code;
@@ -878,29 +879,18 @@ static int s_calculate_in_memory_checksum_helper(
 static int s_calculate_and_add_checksum_to_header_helper(
     struct aws_allocator *allocator,
     struct aws_byte_cursor data,
-    const struct checksum_config_storage *checksum_config,
     struct aws_http_message *out_message,
-    struct aws_byte_buf *out_checksum) {
-    AWS_ASSERT(checksum_config->checksum_algorithm != AWS_SCA_NONE);
+    struct aws_s3_upload_request_checksum_context *checksum_context) {
     AWS_ASSERT(out_message != NULL);
     int ret_code = AWS_OP_ERR;
-
-    struct aws_byte_buf local_encoded_checksum_buf;
-    struct aws_byte_buf *local_encoded_checksum;
-    if (out_checksum == NULL) {
-        local_encoded_checksum = &local_encoded_checksum_buf;
-    } else {
-        local_encoded_checksum = out_checksum;
-    }
-    AWS_ZERO_STRUCT(*local_encoded_checksum);
-    if (s_calculate_in_memory_checksum_helper(allocator, data, checksum_config, local_encoded_checksum)) {
+    if (s_calculate_in_memory_checksum_helper(allocator, data, checksum_context)) {
         goto done;
     }
 
     /* Add the encoded checksum to header. */
     const struct aws_byte_cursor header_name =
-        aws_get_http_header_name_from_checksum_algorithm(checksum_config->checksum_algorithm);
-    struct aws_byte_cursor encoded_checksum_val = aws_byte_cursor_from_buf(local_encoded_checksum);
+        aws_get_http_header_name_from_checksum_algorithm(checksum_context->algorithm);
+    struct aws_byte_cursor encoded_checksum_val = aws_byte_cursor_from_buf(&checksum_context->base64_checksum);
     struct aws_http_headers *headers = aws_http_message_get_headers(out_message);
     if (aws_http_headers_set(headers, header_name, encoded_checksum_val)) {
         goto done;
@@ -908,11 +898,6 @@ static int s_calculate_and_add_checksum_to_header_helper(
 
     ret_code = AWS_OP_SUCCESS;
 done:
-    if (ret_code || out_checksum == NULL) {
-        /* In case of error happen or out_checksum is not set, clean up the encoded checksum. Otherwise, the caller will
-         * own the encoded checksum. */
-        aws_byte_buf_clean_up(local_encoded_checksum);
-    }
     return ret_code;
 }
 
@@ -921,8 +906,7 @@ struct aws_input_stream *aws_s3_message_util_assign_body(
     struct aws_allocator *allocator,
     struct aws_byte_buf *byte_buf,
     struct aws_http_message *out_message,
-    const struct checksum_config_storage *checksum_config,
-    struct aws_byte_buf *out_checksum) {
+    struct aws_s3_upload_request_checksum_context *checksum_context) {
     AWS_PRECONDITION(allocator);
     AWS_PRECONDITION(out_message);
     AWS_PRECONDITION(byte_buf);
@@ -942,8 +926,8 @@ struct aws_input_stream *aws_s3_message_util_assign_body(
         goto error_clean_up;
     }
 
-    if (checksum_config) {
-        if (checksum_config->location == AWS_SCL_TRAILER) {
+    if (checksum_context && checksum_context->algorithm != AWS_SCA_NONE) {
+        if (aws_s3_upload_request_checksum_context_should_add_trailer(checksum_context)) {
             /* aws-chunked encode the payload and add related headers */
 
             /* set Content-Encoding header. If the header already exists, append the existing value to aws-chunked
@@ -975,7 +959,7 @@ struct aws_input_stream *aws_s3_message_util_assign_body(
             if (aws_http_headers_set(
                     headers,
                     g_trailer_header_name,
-                    aws_get_http_header_name_from_checksum_algorithm(checksum_config->checksum_algorithm))) {
+                    aws_get_http_header_name_from_checksum_algorithm(checksum_context->algorithm))) {
                 goto error_clean_up;
             }
             /* set x-amz-decoded-content-length header */
@@ -991,28 +975,28 @@ struct aws_input_stream *aws_s3_message_util_assign_body(
                 goto error_clean_up;
             }
             /* set input stream to chunk stream */
-            struct aws_input_stream *chunk_stream =
-                aws_chunk_stream_new(allocator, input_stream, checksum_config->checksum_algorithm, out_checksum);
+            struct aws_input_stream *chunk_stream = aws_chunk_stream_new(allocator, input_stream, checksum_context);
             if (!chunk_stream) {
                 goto error_clean_up;
             }
             aws_input_stream_release(input_stream);
             input_stream = chunk_stream;
-        } else if (checksum_config->location == AWS_SCL_HEADER) {
+        } else if (aws_s3_upload_request_checksum_context_should_add_header(checksum_context)) {
             /* Calculate the checksum directly from memory and add it to the header. */
             if (s_calculate_and_add_checksum_to_header_helper(
-                    allocator, buffer_byte_cursor, checksum_config, out_message, out_checksum)) {
+                    allocator, buffer_byte_cursor, out_message, checksum_context)) {
                 goto error_clean_up;
             }
-
-        } else if (checksum_config->checksum_algorithm != AWS_SCA_NONE && out_checksum != NULL) {
+        } else if (aws_s3_upload_request_checksum_context_should_calculate(checksum_context)) {
             /* In case checksums still wanted, and we can calculate it directly from the buffer in memory to
              * out_checksum */
-            if (s_calculate_in_memory_checksum_helper(allocator, buffer_byte_cursor, checksum_config, out_checksum)) {
+            if (s_calculate_in_memory_checksum_helper(allocator, buffer_byte_cursor, checksum_context)) {
                 goto error_clean_up;
             }
         }
     }
+
+    /* Set content-length header */
     int64_t stream_length = 0;
     if (aws_input_stream_get_length(input_stream, &stream_length)) {
         goto error_clean_up;
diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
index 7f8015beb..bc3165f15 100644
--- a/tests/CMakeLists.txt
+++ b/tests/CMakeLists.txt
@@ -294,6 +294,11 @@ add_test_case(crc32c_test_invalid_buffer)
 add_test_case(crc32c_test_oneshot)
 add_test_case(crc32c_test_invalid_state)
 
+add_test_case(test_upload_request_checksum_context_get_output_buffer)
+add_test_case(test_upload_request_checksum_context_get_checksum_cursor)
+add_test_case(test_upload_request_checksum_context_error_cases)
+add_test_case(test_upload_request_checksum_context_different_algorithms)
+
 add_net_test_case(verify_checksum_stream)
 add_net_test_case(verify_chunk_stream)
 
diff --git a/tests/mock_s3_server/mock_s3_server.py b/tests/mock_s3_server/mock_s3_server.py
index ba88aee82..c90b1f2c5 100644
--- a/tests/mock_s3_server/mock_s3_server.py
+++ b/tests/mock_s3_server/mock_s3_server.py
@@ -377,25 +377,6 @@ async def send_mock_s3_response(wrapper, request_type, path, generate_body=False
     await send_response_from_json(wrapper, response_file, generate_body=generate_body, generate_body_size=generate_body_size, head_request=head_request)
 
 
-async def maybe_send_error_response(wrapper, exc):
-    if wrapper.conn.our_state not in {h11.IDLE, h11.SEND_RESPONSE}:
-        wrapper.info("...but I can't, because our state is",
-                     wrapper.conn.our_state)
-        return
-    try:
-        await wrapper.send(res)
-    except Exception as e:
-        print(e)
-
-    if not response.head_request:
-        if response.chunked:
-            await wrapper.send(h11.Data(data=b"%X\r\n%s\r\n" % (len(response.data), response.data.encode())))
-        else:
-            await wrapper.send(h11.Data(data=response.data.encode()))
-
-    await wrapper.send(h11.EndOfMessage())
-
-
 def get_request_header_value(request, header_name):
     for header in request.headers:
         if header[0].decode("utf-8").lower() == header_name.lower():
diff --git a/tests/s3_checksum_context_test.c b/tests/s3_checksum_context_test.c
new file mode 100644
index 000000000..647293bd6
--- /dev/null
+++ b/tests/s3_checksum_context_test.c
@@ -0,0 +1,152 @@
+/**
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+
+#include "aws/s3/private/s3_checksum_context.h"
+#include "aws/s3/private/s3_checksums.h"
+#include <aws/common/byte_buf.h>
+#include <aws/testing/aws_test_harness.h>
+
+static int s_test_upload_request_checksum_context_get_output_buffer(struct aws_allocator *allocator, void *ctx) {
+    (void)ctx;
+
+    struct aws_s3_meta_request_checksum_config_storage config = {
+        .allocator = allocator,
+        .checksum_algorithm = AWS_SCA_CRC32,
+        .location = AWS_SCL_HEADER,
+        .has_full_object_checksum = false,
+    };
+    AWS_ZERO_STRUCT(config.full_object_checksum);
+
+    /* Test get output buffer with valid context */
+    struct aws_s3_upload_request_checksum_context *context =
+        aws_s3_upload_request_checksum_context_new(allocator, &config);
+    ASSERT_NOT_NULL(context);
+
+    struct aws_byte_buf *output_buffer = aws_s3_upload_request_checksum_context_get_output_buffer(context);
+    ASSERT_NOT_NULL(output_buffer);
+    ASSERT_TRUE(output_buffer->capacity > 0);
+
+    aws_s3_upload_request_checksum_context_release(context);
+
+    /* Test get output buffer with NULL context */
+    output_buffer = aws_s3_upload_request_checksum_context_get_output_buffer(NULL);
+    ASSERT_NULL(output_buffer);
+
+    return AWS_OP_SUCCESS;
+}
+AWS_TEST_CASE(
+    test_upload_request_checksum_context_get_output_buffer,
+    s_test_upload_request_checksum_context_get_output_buffer)
+
+static int s_test_upload_request_checksum_context_get_checksum_cursor(struct aws_allocator *allocator, void *ctx) {
+    (void)ctx;
+
+    struct aws_s3_meta_request_checksum_config_storage config = {
+        .allocator = allocator,
+        .checksum_algorithm = AWS_SCA_CRC32,
+        .location = AWS_SCL_HEADER,
+        .has_full_object_checksum = false,
+    };
+    AWS_ZERO_STRUCT(config.full_object_checksum);
+
+    /* Test get checksum cursor with context that has no calculated checksum */
+    struct aws_s3_upload_request_checksum_context *context =
+        aws_s3_upload_request_checksum_context_new(allocator, &config);
+    ASSERT_NOT_NULL(context);
+
+    struct aws_byte_cursor cursor = aws_s3_upload_request_checksum_context_get_checksum_cursor(context);
+    ASSERT_TRUE(cursor.len == 0);
+    ASSERT_NULL(cursor.ptr);
+
+    aws_s3_upload_request_checksum_context_release(context);
+
+    /* Test get checksum cursor with context that has calculated checksum */
+    struct aws_byte_cursor existing_checksum = aws_byte_cursor_from_c_str("dGVzdA==");
+    context =
+        aws_s3_upload_request_checksum_context_new_with_existing_base64_checksum(allocator, &config, existing_checksum);
+    ASSERT_NOT_NULL(context);
+
+    cursor = aws_s3_upload_request_checksum_context_get_checksum_cursor(context);
+    ASSERT_TRUE(cursor.len == existing_checksum.len);
+    ASSERT_TRUE(aws_byte_cursor_eq(&cursor, &existing_checksum));
+
+    aws_s3_upload_request_checksum_context_release(context);
+
+    /* Test get checksum cursor with NULL context */
+    cursor = aws_s3_upload_request_checksum_context_get_checksum_cursor(NULL);
+    ASSERT_TRUE(cursor.len == 0);
+    ASSERT_NULL(cursor.ptr);
+
+    return AWS_OP_SUCCESS;
+}
+AWS_TEST_CASE(
+    test_upload_request_checksum_context_get_checksum_cursor,
+    s_test_upload_request_checksum_context_get_checksum_cursor)
+
+static int s_test_upload_request_checksum_context_error_cases(struct aws_allocator *allocator, void *ctx) {
+    (void)ctx;
+
+    struct aws_s3_meta_request_checksum_config_storage config = {
+        .allocator = allocator,
+        .checksum_algorithm = AWS_SCA_CRC32,
+        .location = AWS_SCL_HEADER,
+        .has_full_object_checksum = false,
+    };
+    AWS_ZERO_STRUCT(config.full_object_checksum);
+
+    /* Test creation with mismatched checksum size */
+    struct aws_byte_cursor wrong_size_checksum = aws_byte_cursor_from_c_str("short");
+    struct aws_s3_upload_request_checksum_context *context =
+        aws_s3_upload_request_checksum_context_new_with_existing_base64_checksum(
+            allocator, &config, wrong_size_checksum);
+    ASSERT_NULL(context);
+
+    /* Test helper functions with NULL context */
+    ASSERT_FALSE(aws_s3_upload_request_checksum_context_should_calculate(NULL));
+    ASSERT_FALSE(aws_s3_upload_request_checksum_context_should_add_header(NULL));
+    ASSERT_FALSE(aws_s3_upload_request_checksum_context_should_add_trailer(NULL));
+
+    /* Test acquire/release with NULL context */
+    ASSERT_NULL(aws_s3_upload_request_checksum_context_acquire(NULL));
+    ASSERT_NULL(aws_s3_upload_request_checksum_context_release(NULL));
+
+    return AWS_OP_SUCCESS;
+}
+AWS_TEST_CASE(test_upload_request_checksum_context_error_cases, s_test_upload_request_checksum_context_error_cases)
+
+static int s_test_upload_request_checksum_context_different_algorithms(struct aws_allocator *allocator, void *ctx) {
+    (void)ctx;
+
+    /* Test different checksum algorithms */
+    enum aws_s3_checksum_algorithm algorithms[] = {
+        AWS_SCA_CRC32, AWS_SCA_CRC32C, AWS_SCA_SHA1, AWS_SCA_SHA256, AWS_SCA_CRC64NVME};
+
+    for (size_t i = 0; i < AWS_ARRAY_SIZE(algorithms); ++i) {
+        struct aws_s3_meta_request_checksum_config_storage config = {
+            .allocator = allocator,
+            .checksum_algorithm = algorithms[i],
+            .location = AWS_SCL_HEADER,
+            .has_full_object_checksum = false,
+        };
+        AWS_ZERO_STRUCT(config.full_object_checksum);
+
+        struct aws_s3_upload_request_checksum_context *context =
+            aws_s3_upload_request_checksum_context_new(allocator, &config);
+        ASSERT_NOT_NULL(context);
+        ASSERT_INT_EQUALS(algorithms[i], context->algorithm);
+        ASSERT_INT_EQUALS(AWS_SCL_HEADER, context->location);
+        ASSERT_TRUE(context->encoded_checksum_size > 0);
+        ASSERT_TRUE(aws_s3_upload_request_checksum_context_should_calculate(context));
+        ASSERT_TRUE(aws_s3_upload_request_checksum_context_should_add_header(context));
+        ASSERT_FALSE(aws_s3_upload_request_checksum_context_should_add_trailer(context));
+
+        aws_s3_upload_request_checksum_context_release(context);
+    }
+
+    return AWS_OP_SUCCESS;
+}
+AWS_TEST_CASE(
+    test_upload_request_checksum_context_different_algorithms,
+    s_test_upload_request_checksum_context_different_algorithms)
diff --git a/tests/s3_checksum_stream_test.c b/tests/s3_checksum_stream_test.c
index e4fb3eac7..76e63529f 100644
--- a/tests/s3_checksum_stream_test.c
+++ b/tests/s3_checksum_stream_test.c
@@ -3,6 +3,7 @@
  * SPDX-License-Identifier: Apache-2.0.
  */
 
+#include "aws/s3/private/s3_checksum_context.h"
 #include "aws/s3/private/s3_checksums.h"
 #include "s3_tester.h"
 #include <aws/common/byte_buf.h>
@@ -111,9 +112,27 @@ static int s_stream_chunk(
     struct aws_byte_buf *output,
     enum aws_s3_checksum_algorithm algorithm,
     struct aws_byte_buf *checksum_result) {
+
+    /* Create checksum config for the context */
+    struct aws_s3_meta_request_checksum_config_storage config = {
+        .allocator = allocator,
+        .checksum_algorithm = algorithm,
+        .location = AWS_SCL_TRAILER,
+        .has_full_object_checksum = false,
+    };
+    AWS_ZERO_STRUCT(config.full_object_checksum);
+
+    /* Create checksum context */
+    struct aws_s3_upload_request_checksum_context *context =
+        aws_s3_upload_request_checksum_context_new(allocator, &config);
+    if (!context) {
+        return AWS_OP_ERR;
+    }
+
     struct aws_input_stream *cursor_stream = aws_input_stream_new_from_cursor(allocator, input);
-    struct aws_input_stream *stream = aws_chunk_stream_new(allocator, cursor_stream, algorithm, checksum_result);
+    struct aws_input_stream *stream = aws_chunk_stream_new(allocator, cursor_stream, context);
     aws_input_stream_release(cursor_stream);
+
     struct aws_stream_status status;
     AWS_ZERO_STRUCT(status);
     while (!status.is_end_of_stream) {
@@ -123,7 +142,19 @@ static int s_stream_chunk(
         read_buf->len = 0;
         ASSERT_TRUE(aws_input_stream_get_status(stream, &status) == 0);
     }
+
+    /* Copy checksum result if requested */
+    if (checksum_result) {
+        struct aws_byte_cursor checksum_cursor = aws_s3_upload_request_checksum_context_get_checksum_cursor(context);
+        if (aws_byte_buf_init_copy_from_cursor(checksum_result, allocator, checksum_cursor) != AWS_OP_SUCCESS) {
+            aws_input_stream_release(stream);
+            aws_s3_upload_request_checksum_context_release(context);
+            return AWS_OP_ERR;
+        }
+    }
+
     aws_input_stream_release(stream);
+    aws_s3_upload_request_checksum_context_release(context);
     return AWS_OP_SUCCESS;
 }
 
@@ -143,6 +174,7 @@ static int compare_chunk_stream(
     struct aws_byte_buf read_buf;
     aws_byte_buf_init(&read_buf, allocator, buffer_size);
     for (size_t i = 0; i < AWS_ARRAY_SIZE(s_checksum_algo_priority_list); i++) {
+        AWS_ZERO_STRUCT(streamed_encoded_checksum);
         enum aws_s3_checksum_algorithm algorithm = s_checksum_algo_priority_list[i];
         aws_base64_compute_encoded_len(aws_get_digest_size_from_checksum_algorithm(algorithm), &encoded_len);
         size_t total_len =
diff --git a/tests/s3_data_plane_tests.c b/tests/s3_data_plane_tests.c
index 85741084f..53bfc1bf9 100644
--- a/tests/s3_data_plane_tests.c
+++ b/tests/s3_data_plane_tests.c
@@ -3647,7 +3647,7 @@ static int s_test_s3_upload_part_message_helper(struct aws_allocator *allocator,
     struct aws_string *upload_id = aws_string_new_from_c_str(allocator, "dummy_upload_id");
 
     struct aws_http_message *new_message = aws_s3_upload_part_message_new(
-        allocator, base_message, &test_buffer, part_number, upload_id, should_compute_content_md5, NULL, NULL);
+        allocator, base_message, &test_buffer, part_number, upload_id, should_compute_content_md5, NULL);
 
     struct aws_http_headers *new_headers = aws_http_message_get_headers(new_message);
     if (should_compute_content_md5) {
diff --git a/tests/s3_mock_server_tests.c b/tests/s3_mock_server_tests.c
index 9fd1e44a7..b7b85f624 100644
--- a/tests/s3_mock_server_tests.c
+++ b/tests/s3_mock_server_tests.c
@@ -434,12 +434,23 @@ TEST_CASE(multipart_upload_with_network_interface_names_mock_server) {
 #endif
 }
 
+/* Total hack to flip the bytes. */
+static void s_after_prepare_upload_part_finish(struct aws_s3_request *request) {
+    if (request->num_times_prepared > 1) {
+        /* mock that the body buffer was messed up in memory */
+        request->request_body.buffer[1]++;
+    }
+}
+
+/**
+ * This test is built for
+ * 1. We had a memory leak when the retry was triggered and the checksum was calculated.
+ *      The retry will initialize the checksum buffer again, but the previous one was not freed.
+ * 2. We had a bug where the retry will mangle the data with the error response from server.
+ * 3. Don't recalculate the checksum when retrying.
+ */
 TEST_CASE(multipart_upload_checksum_with_retry_mock_server) {
     (void)ctx;
-    /**
-     * We had a memory leak when the retry was triggered and the checksum was calculated.
-     * The retry will initialize the checksum buffer again, but the previous one was not freed.
-     */
     struct aws_s3_tester tester;
     ASSERT_SUCCESS(aws_s3_tester_init(allocator, &tester));
     struct aws_s3_tester_client_options client_options = {
@@ -449,37 +460,73 @@ TEST_CASE(multipart_upload_checksum_with_retry_mock_server) {
 
     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");
-
-    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,
-            },
-        .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]));
-
+    {
+        /* 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,
+                },
+            .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);
-    aws_s3_meta_request_test_results_clean_up(&meta_request_test_results);
 
     return AWS_OP_SUCCESS;
 }
diff --git a/tests/s3_request_messages_tests.c b/tests/s3_request_messages_tests.c
index eef4d4cf2..1cfdffb8e 100644
--- a/tests/s3_request_messages_tests.c
+++ b/tests/s3_request_messages_tests.c
@@ -524,8 +524,7 @@ static int s_test_s3_message_util_assign_body(struct aws_allocator *allocator, v
     struct aws_byte_buf test_buffer;
     ASSERT_SUCCESS(s_fill_byte_buf(&test_buffer, allocator, test_buffer_size));
 
-    struct aws_input_stream *input_stream =
-        aws_s3_message_util_assign_body(allocator, &test_buffer, message, NULL, NULL);
+    struct aws_input_stream *input_stream = aws_s3_message_util_assign_body(allocator, &test_buffer, message, NULL);
     ASSERT_TRUE(input_stream != NULL);
 
     ASSERT_TRUE(aws_http_message_get_body_stream(message) == input_stream);
@@ -733,8 +732,8 @@ static int s_test_s3_upload_part_message_new(struct aws_allocator *allocator, vo
 
     struct aws_string *upload_id = aws_string_new_from_c_str(allocator, UPLOAD_ID);
 
-    struct aws_http_message *upload_part_message = aws_s3_upload_part_message_new(
-        allocator, original_message, &part_buffer, PART_NUMBER, upload_id, false, NULL, NULL);
+    struct aws_http_message *upload_part_message =
+        aws_s3_upload_part_message_new(allocator, original_message, &part_buffer, PART_NUMBER, upload_id, false, NULL);
     ASSERT_TRUE(upload_part_message != NULL);
 
     ASSERT_SUCCESS(s_test_http_message_request_method(upload_part_message, "PUT"));
@@ -784,8 +783,8 @@ static int s_test_s3_upload_part_message_fail(struct aws_allocator *allocator, v
 
     struct aws_string *upload_id = aws_string_new_from_c_str(allocator, UPLOAD_ID);
 
-    struct aws_http_message *upload_part_message = aws_s3_upload_part_message_new(
-        allocator, original_message, &part_buffer, PART_NUMBER, upload_id, false, NULL, NULL);
+    struct aws_http_message *upload_part_message =
+        aws_s3_upload_part_message_new(allocator, original_message, &part_buffer, PART_NUMBER, upload_id, false, NULL);
     ASSERT_NULL(upload_part_message);
 
     aws_string_destroy(upload_id);