fix broken protocol tests; rework codegen hierarchy

runtime/protocol/http-client-engines/http-client-engine-ktor/jvm/src/aws/smithy/kotlin/runtime/http/engine/ktor/KtorUtils.kt

@@ -97,6 +97,7 @@ internal class KtorContentStream(private val channel: ByteReadChannel) : SdkByte
 // wrapper around a ByteReadChannel that implements the content as an SDK (streaming) HttpBody
 internal class KtorHttpBody(channel: ByteReadChannel) : HttpBody.Streaming() {
     private val source = KtorContentStream(channel)
+    override fun deepCopy(): HttpBody = throwSingleConsumptionException()
     override fun readFrom(): SdkByteReadChannel = source
 }
 

runtime/protocol/http/common/src/aws/smithy/kotlin/runtime/http/Headers.kt

@@ -4,8 +4,7 @@
  */
 package aws.smithy.kotlin.runtime.http
 
-import aws.smithy.kotlin.runtime.http.util.StringValuesMap
-import aws.smithy.kotlin.runtime.http.util.StringValuesMapBuilder
+import aws.smithy.kotlin.runtime.http.util.*
 import aws.smithy.kotlin.runtime.http.util.StringValuesMapImpl
 
 /**
@@ -35,13 +34,12 @@ private object EmptyHeaders : Headers {
 /**
  * Build an immutable HTTP header map
  */
-class HeadersBuilder : StringValuesMapBuilder(true, 8) {
+class HeadersBuilder : StringValuesMapBuilder(true, 8), CanDeepCopy<HeadersBuilder> {
     override fun toString(): String = "HeadersBuilder ${entries()} "
-    override fun build(): Headers {
-        require(!built) { "HeadersBuilder can only build a single Headers instance" }
-        built = true
-        return HeadersImpl(values)
-    }
+    override fun build(): Headers = HeadersImpl(values)
+
+    override fun deepCopy(): HeadersBuilder =
+        HeadersBuilder().apply { values.putAll(this@HeadersBuilder.values.deepCopy()) }
 }
 
 private class HeadersImpl(

runtime/protocol/http/common/src/aws/smithy/kotlin/runtime/http/HttpBody.kt

@@ -5,12 +5,17 @@
 package aws.smithy.kotlin.runtime.http
 
 import aws.smithy.kotlin.runtime.content.ByteStream
+import aws.smithy.kotlin.runtime.http.util.CanDeepCopy
 import aws.smithy.kotlin.runtime.io.SdkByteReadChannel
 
 /**
  * HTTP payload to be sent to a peer
  */
-sealed class HttpBody {
+sealed class HttpBody : CanDeepCopy<HttpBody> {
+    /**
+     * Flag indicating the body can be consumed multiple times.
+     */
+    open val isReplayable: Boolean = true
 
     /**
      * Specifies the length of this [HttpBody] content
@@ -23,6 +28,7 @@ sealed class HttpBody {
      */
     object Empty : HttpBody() {
         override val contentLength: Long = 0
+        override fun deepCopy(): HttpBody = this // Deep copies are unnecessary for empty bodies
     }
 
     /**
@@ -35,6 +41,12 @@ sealed class HttpBody {
          * Provides [ByteArray] to be sent to peer
          */
         abstract fun bytes(): ByteArray
+
+        override fun deepCopy(): Bytes = object : Bytes() {
+            private val copiedBytes = this@Bytes.bytes().copyOf()
+            override fun bytes(): ByteArray = copiedBytes
+            override val contentLength: Long? = copiedBytes.size.toLong()
+        }
     }
 
     /**
@@ -52,14 +64,20 @@ sealed class HttpBody {
          * Flag indicating the stream can be consumed multiple times. If `false` [reset] will throw an
          * [UnsupportedOperationException].
          */
-        open val isReplayable: Boolean = false
+        override val isReplayable: Boolean = false
 
         /**
          * Reset the stream such that the next call to [readFrom] provides a fresh channel.
          * @throws UnsupportedOperationException if the stream can only be consumed once. Consumers can check
          * [isReplayable] before calling
          */
-        open fun reset() { throw UnsupportedOperationException("${this::class.simpleName} can only be consumed once") }
+        open fun reset() { throwSingleConsumptionException() }
+
+        /**
+         * Throw a general exception upon attempting to consume the stream more than once.
+         */
+        protected fun throwSingleConsumptionException(): Nothing =
+            throw UnsupportedOperationException("${this::class.simpleName} can only be consumed once")
     }
 }
 
@@ -73,11 +91,13 @@ fun ByteStream.toHttpBody(): HttpBody = when (val byteStream = this) {
     }
     is ByteStream.OneShotStream -> object : HttpBody.Streaming() {
         override val contentLength: Long? = byteStream.contentLength
+        override fun deepCopy(): Streaming = throwSingleConsumptionException()
         override fun readFrom(): SdkByteReadChannel = byteStream.readFrom()
     }
     is ByteStream.ReplayableStream -> object : HttpBody.Streaming() {
         private var channel: SdkByteReadChannel? = null
         override val contentLength: Long? = byteStream.contentLength
+        override fun deepCopy(): Streaming = this // Replayable streams copy themselves by default
         override fun readFrom(): SdkByteReadChannel = channel ?: byteStream.newReader().also { channel = it }
         override val isReplayable: Boolean = true
         override fun reset() {

runtime/protocol/http/common/src/aws/smithy/kotlin/runtime/http/QueryParameters.kt

@@ -4,9 +4,7 @@
  */
 package aws.smithy.kotlin.runtime.http
 
-import aws.smithy.kotlin.runtime.http.util.StringValuesMap
-import aws.smithy.kotlin.runtime.http.util.StringValuesMapBuilder
-import aws.smithy.kotlin.runtime.http.util.StringValuesMapImpl
+import aws.smithy.kotlin.runtime.http.util.*
 import aws.smithy.kotlin.runtime.util.text.urlEncodeComponent
 
 /**
@@ -33,13 +31,12 @@ private object EmptyQueryParameters : QueryParameters {
     override fun isEmpty(): Boolean = true
 }
 
-class QueryParametersBuilder : StringValuesMapBuilder(true, 8) {
+class QueryParametersBuilder : StringValuesMapBuilder(true, 8), CanDeepCopy<QueryParametersBuilder> {
     override fun toString(): String = "QueryParametersBuilder ${entries()} "
-    override fun build(): QueryParameters {
-        require(!built) { "QueryParametersBuilder can only build a single instance" }
-        built = true
-        return QueryParametersImpl(values)
-    }
+    override fun build(): QueryParameters = QueryParametersImpl(values)
+
+    override fun deepCopy(): QueryParametersBuilder =
+        QueryParametersBuilder().apply { values.putAll(this@QueryParametersBuilder.values.deepCopy()) }
 }
 
 fun Map<String, String>.toQueryParameters(): QueryParameters {

runtime/protocol/http/common/src/aws/smithy/kotlin/runtime/http/Url.kt

@@ -4,6 +4,7 @@
  */
 package aws.smithy.kotlin.runtime.http
 
+import aws.smithy.kotlin.runtime.http.util.CanDeepCopy
 import aws.smithy.kotlin.runtime.util.InternalApi
 import aws.smithy.kotlin.runtime.util.text.encodeUrlPath
 
@@ -105,7 +106,7 @@ data class UserInfo(val username: String, val password: String)
 /**
  * Construct a URL by it's individual components
  */
-class UrlBuilder {
+class UrlBuilder : CanDeepCopy<UrlBuilder> {
     var scheme = Protocol.HTTPS
     var host: String = ""
     var port: Int? = null
@@ -130,6 +131,17 @@ class UrlBuilder {
         forceQuery
     )
 
+    override fun deepCopy(): UrlBuilder = UrlBuilder().apply {
+        scheme = this@UrlBuilder.scheme
+        host = this@UrlBuilder.host
+        port = this@UrlBuilder.port
+        path = this@UrlBuilder.path
+        parameters = this@UrlBuilder.parameters.deepCopy()
+        fragment = this@UrlBuilder.fragment
+        userInfo = this@UrlBuilder.userInfo?.copy()
+        forceQuery = this@UrlBuilder.forceQuery
+    }
+
     override fun toString(): String =
         "UrlBuilder(scheme=$scheme, host='$host', port=$port, path='$path', parameters=$parameters, fragment=$fragment, userInfo=$userInfo, forceQuery=$forceQuery)"
 }

runtime/protocol/http/common/src/aws/smithy/kotlin/runtime/http/retries/StandardRetryFeature.kt → runtime/protocol/http/common/src/aws/smithy/kotlin/runtime/http/middleware/RetryFeature.kt

@@ -3,35 +3,44 @@
  * SPDX-License-Identifier: Apache-2.0.
  */
 
-package aws.smithy.kotlin.runtime.http.retries
+package aws.smithy.kotlin.runtime.http.middleware
 
 import aws.smithy.kotlin.runtime.http.Feature
 import aws.smithy.kotlin.runtime.http.FeatureKey
 import aws.smithy.kotlin.runtime.http.HttpClientFeatureFactory
 import aws.smithy.kotlin.runtime.http.operation.SdkHttpOperation
+import aws.smithy.kotlin.runtime.http.operation.deepCopy
 import aws.smithy.kotlin.runtime.retries.RetryPolicy
 import aws.smithy.kotlin.runtime.retries.RetryStrategy
 
-class StandardRetryFeature(private val strategy: RetryStrategy, private val policy: RetryPolicy<Any?>) : Feature {
+class RetryFeature(private val strategy: RetryStrategy, private val policy: RetryPolicy<Any?>) : Feature {
     class Config {
         var strategy: RetryStrategy? = null
         var policy: RetryPolicy<Any?>? = null
     }
 
-    companion object Feature : HttpClientFeatureFactory<Config, StandardRetryFeature> {
-        override val key: FeatureKey<StandardRetryFeature> = FeatureKey("StandardRetryFeature")
+    companion object Feature : HttpClientFeatureFactory<Config, RetryFeature> {
+        override val key: FeatureKey<RetryFeature> = FeatureKey("RetryFeature")
 
-        override fun create(block: Config.() -> Unit): StandardRetryFeature {
+        override fun create(block: Config.() -> Unit): RetryFeature {
             val config = Config().apply(block)
             val strategy = requireNotNull(config.strategy) { "strategy is required" }
             val policy = requireNotNull(config.policy) { "policy is required" }
-            return StandardRetryFeature(strategy, policy)
+            return RetryFeature(strategy, policy)
         }
     }
 
     override fun <I, O> install(operation: SdkHttpOperation<I, O>) {
         operation.execution.finalize.intercept { req, next ->
-            strategy.retry(policy) { next.call(req) }
+            if (req.subject.isRetryable) {
+                strategy.retry(policy) {
+                    // Deep copy the request because later middlewares (e.g., signing) mutate it
+                    val reqCopy = req.deepCopy()
+                    next.call(reqCopy)
+                }
+            } else {
+                next.call(req)
+            }
         }
     }
 }

runtime/protocol/http/common/src/aws/smithy/kotlin/runtime/http/operation/OperationRequest.kt

@@ -6,6 +6,7 @@
 package aws.smithy.kotlin.runtime.http.operation
 
 import aws.smithy.kotlin.runtime.client.ExecutionContext
+import aws.smithy.kotlin.runtime.http.util.CanDeepCopy
 
 /**
  * Wrapper around a type [subject] with an execution context.
@@ -17,3 +18,11 @@ import aws.smithy.kotlin.runtime.client.ExecutionContext
  * @param subject The input type
  */
 data class OperationRequest<T>(val context: ExecutionContext, val subject: T)
+
+/**
+ * Deep copy an [OperationRequest] to a new request. Note that, because [context] is...well, context...it's considered
+ * transient and is not part of the copy. The subject itself, however, is deeply copied.
+ * @return A new [OperationRequest] with the same context and a deeply-copied subject.
+ */
+fun <T : CanDeepCopy<T>> OperationRequest<T>.deepCopy(): OperationRequest<T> =
+    OperationRequest(context, subject.deepCopy())

runtime/protocol/http/common/src/aws/smithy/kotlin/runtime/http/request/HttpRequestBuilder.kt

@@ -6,35 +6,37 @@ package aws.smithy.kotlin.runtime.http.request
 
 import aws.smithy.kotlin.runtime.http.*
 import aws.smithy.kotlin.runtime.http.content.ByteArrayContent
+import aws.smithy.kotlin.runtime.http.util.CanDeepCopy
 import aws.smithy.kotlin.runtime.io.*
 import aws.smithy.kotlin.runtime.util.InternalApi
 
 /**
  * Used to construct an HTTP request
+ * @param method The HTTP method (verb) to use when making the request
+ * @param url Endpoint to make request to
+ * @param headers HTTP headers
+ * @param body Outgoing payload. Initially empty
  */
-class HttpRequestBuilder {
+class HttpRequestBuilder private constructor(
+    var method: HttpMethod,
+    val url: UrlBuilder,
+    val headers: HeadersBuilder,
+    var body: HttpBody,
+) : CanDeepCopy<HttpRequestBuilder> {
     /**
-     * The HTTP method (verb) to use when making the request
+     * Indicates whether this HTTP request could be retried. Some requests with streaming bodies are unsuitable for
+     * retries.
      */
-    var method: HttpMethod = HttpMethod.GET
+    val isRetryable: Boolean
+        get() = body.isReplayable
 
-    /**
-     * Endpoint to make request to
-     */
-    val url: UrlBuilder = UrlBuilder()
-
-    /**
-     * HTTP headers
-     */
-    val headers: HeadersBuilder = HeadersBuilder()
-
-    /**
-     * Outgoing payload. Initially empty
-     */
-    var body: HttpBody = HttpBody.Empty
+    constructor() : this(HttpMethod.GET, UrlBuilder(), HeadersBuilder(), HttpBody.Empty)
 
     fun build(): HttpRequest = HttpRequest(method, url.build(), if (headers.isEmpty()) Headers.Empty else headers.build(), body)
 
+    override fun deepCopy(): HttpRequestBuilder =
+        HttpRequestBuilder(method, url.deepCopy(), headers.deepCopy(), body.deepCopy())
+
     override fun toString(): String = buildString {
         append("HttpRequestBuilder(method=$method, url=$url, headers=$headers, body=$body)")
     }

runtime/protocol/http/common/src/aws/smithy/kotlin/runtime/http/util/CanDeepCopy.kt

@@ -0,0 +1,10 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+
+package aws.smithy.kotlin.runtime.http.util
+
+interface CanDeepCopy<T> {
+    fun deepCopy(): T
+}

runtime/protocol/http/common/src/aws/smithy/kotlin/runtime/http/util/StringValuesMap.kt

@@ -64,8 +64,10 @@ internal open class StringValuesMapImpl(
     override val caseInsensitiveName: Boolean = false,
     initialValues: Map<String, List<String>> = emptyMap()
 ) : StringValuesMap {
-    protected val values: Map<String, List<String>> by lazy {
-        if (caseInsensitiveName) CaseInsensitiveMap<List<String>>().apply { putAll(initialValues) } else values.toMap()
+    protected val values: Map<String, List<String>> = run {
+        // Make a defensive copy so modifications to the initialValues don't mutate our internal copy
+        val copiedValues = initialValues.deepCopy()
+        if (caseInsensitiveName) CaseInsensitiveMap<List<String>>().apply { putAll(copiedValues) } else copiedValues
     }
 
     override fun getAll(name: String): List<String>? = values[name]
@@ -81,13 +83,17 @@ internal open class StringValuesMapImpl(
     override fun isEmpty(): Boolean = values.isEmpty()
 }
 
+/**
+ * Perform a deep copy of this map, specifically duplicating the value lists so that they're insulated from changes.
+ * @return A new map instance with copied value lists.
+ */
+internal fun Map<String, List<String>>.deepCopy() = mapValues { (_, v) -> v.toMutableList() }
+
 @InternalApi
 open class StringValuesMapBuilder(val caseInsensitiveName: Boolean = false, size: Int = 8) {
     protected val values: MutableMap<String, MutableList<String>> =
         if (caseInsensitiveName) CaseInsensitiveMap() else LinkedHashMap(size)
 
-    protected var built = false
-
     fun getAll(name: String): List<String>? = values[name]
 
     operator fun contains(name: String): Boolean = name in values
@@ -154,16 +160,8 @@ open class StringValuesMapBuilder(val caseInsensitiveName: Boolean = false, size
 
     fun clear() = values.clear()
 
-    open fun build(): StringValuesMap {
-        require(!built) { "StringValuesMapBuilder can only build a single instance" }
-        built = true
-        return StringValuesMapImpl(caseInsensitiveName, values)
-    }
+    open fun build(): StringValuesMap = StringValuesMapImpl(caseInsensitiveName, values)
 
-    private fun ensureListForKey(name: String, size: Int): MutableList<String> {
-        if (built) {
-            throw IllegalStateException("Cannot modify a builder when final structure has already been built")
-        }
-        return values[name] ?: ArrayList<String>(size).also { values[name] = it }
-    }
+    private fun ensureListForKey(name: String, size: Int): MutableList<String> =
+        values[name] ?: ArrayList<String>(size).also { values[name] = it }
 }