Fix Content-Length and retries for seekable streaming payloads

packages/smithy-core/.changes/next-release/smithy-core-enhancement-ce57b5cfed2740908d9e5222d3f58f47.json

@@ -0,0 +1,4 @@
+{
+  "type": "enhancement",
+  "description": "Added retries for requests with a seekable streaming payload, which are resent from the position the stream was at when the operation was called."
+}

packages/smithy-core/src/smithy_core/aio/client.py

@@ -30,7 +30,7 @@
 from ..interfaces.auth import AuthOption, AuthSchemeResolver
 from ..schemas import APIOperation
 from ..serializers import SerializeableShape
-from ..shapes import ShapeID
+from ..shapes import ShapeID, ShapeType
 from ..types import PropertyKey
 from ..types import TypedProperties as _TypedProperties
 from .eventstream import DuplexEventStream, InputEventStream, OutputEventStream
@@ -92,8 +92,26 @@ class ClientCall[I: SerializeableShape, O: DeserializeableShape]:
     """The retry scope for the operation."""
 
     def retryable(self) -> bool:
-        # TODO: check to see if the stream is seekable
-        return self.operation.input_stream_member is None
+        """Whether the operation can be retried.
+
+        Operations with an input event stream are never retried. A request for any
+        other operation is retried only if its body can be rewound to where it started.
+        """
+        member = self.operation.input_stream_member
+        return member is None or member.shape_type is not ShapeType.UNION
+
+
+async def _rewind_position(body: Any) -> int | None:
+    """Find where a request body is rewound to before it's resent.
+
+    :returns: The body's current position, or None if it can't be rewound.
+    """
+    if body is None or isinstance(body, bytes | bytearray):
+        return 0
+    try:
+        return await seek(body, 0, 1)
+    except (OSError, ValueError, NotImplementedError):
+        return None
 
 
 class RequestPipeline[TRequest: Request, TResponse: Response]:
@@ -382,7 +400,8 @@ async def _retry[I: SerializeableShape, O: DeserializeableShape](
         request_context: RequestContext[I, TRequest],
         request_future: Future[RequestContext[I, TRequest]] | None,
     ) -> OutputContext[I, O, TRequest | None, TResponse | None]:
-        if not call.retryable():
+        body = request_context.transport_request.body
+        if not call.retryable() or (start := await _rewind_position(body)) is None:
             return await self._send_with_rate_limiting(
                 call, request_context, request_future
             )
@@ -444,7 +463,11 @@ async def _retry[I: SerializeableShape, O: DeserializeableShape](
                     retry_token.retry_delay,
                 )
 
-                await seek(request_context.transport_request.body, 0)
+                try:
+                    await seek(body, start)
+                except (OSError, ValueError, NotImplementedError) as e:
+                    output_context.response.add_note(f"Retry not attempted: {e}")
+                    return output_context
             else:
                 await retry_strategy.record_success(token=retry_token)
                 return output_context

packages/smithy-core/tests/unit/aio/_pipeline_harness.py

@@ -12,6 +12,7 @@
 from smithy_core.documents import TypeRegistry
 from smithy_core.endpoints import EndpointResolverParams
 from smithy_core.interceptors import InterceptorChain
+from smithy_core.prelude import BLOB
 from smithy_core.response import EMPTY_RESPONSE_METADATA, ResponseMetadata
 from smithy_core.schemas import APIOperation, Schema
 from smithy_core.serializers import ShapeSerializer
@@ -86,7 +87,29 @@ def deserialize(cls, deserializer: ShapeDeserializer) -> Self:
     effective_auth_schemes=[],
 )
 
-# ``ClientCall.retryable()`` is False for operations with a streaming input, so
+_DATA_STREAM_INPUT_SCHEMA = Schema.collection(
+    id=ShapeID("com.example#DataStreamInput"),
+    members={"body": {"target": BLOB, "traits": [StreamingTrait()]}},
+)
+
+# An operation whose input is a data stream rather than an event stream.
+DATA_STREAM_OPERATION = APIOperation(
+    input=StubInput,
+    output=StubOutput,
+    static_schema=(
+        ShapeType.OPERATION.value,
+        "com.example",
+        "DataStreamOperation",
+        0,
+        _DATA_STREAM_INPUT_SCHEMA,
+        _UNARY_OUTPUT_SCHEMA,
+        (),
+    ),
+    error_registry=TypeRegistry({}),
+    effective_auth_schemes=[],
+)
+
+# ``ClientCall.retryable()`` is False for operations with an input event stream, so
 # OPERATION above never enters the retry loop. This one has no input stream and so
 # exercises it.
 UNARY_OPERATION = APIOperation(

packages/smithy-core/tests/unit/aio/test_client.py

@@ -1,15 +1,18 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0
 
+from collections.abc import AsyncIterator, Callable
 from dataclasses import replace
+from io import BytesIO, UnsupportedOperation
 from typing import Any, cast
 from unittest.mock import AsyncMock, Mock, patch
 
 import pytest
 from smithy_core.aio.client import AsyncClient
 from smithy_core.aio.eventstream import DuplexEventStream, InputEventStream
 from smithy_core.aio.interfaces import ClientTransport
-from smithy_core.aio.retries import AdaptiveRetryStrategy
+from smithy_core.aio.retries import AdaptiveRetryStrategy, SimpleRetryStrategy
+from smithy_core.aio.types import AsyncBytesReader
 from smithy_core.auth import AuthOption
 from smithy_core.exceptions import CallError, SmithyError, UnsupportedTransportError
 from smithy_core.interceptors import (
@@ -19,17 +22,21 @@
     OutputContext,
 )
 from smithy_core.response import EMPTY_RESPONSE_METADATA
+from smithy_core.retries import ExponentialRetryBackoffStrategy
 from smithy_core.shapes import ShapeID
 from smithy_core.types import TypedProperties
 
 from ._pipeline_harness import (
+    DATA_STREAM_OPERATION,
+    OPERATION,
     DuplexTransport,
     NonDuplexTransport,
     StubEndpointResolver,
     StubEvent,
     StubEventReceiver,
     StubInput,
     StubOutput,
+    StubRequest,
     UndeclaredTransport,
     client_call,
     pipeline_harness,
@@ -319,3 +326,116 @@ async def test_aexit_closes_client() -> None:
 
     assert client.closed is True
     assert client.transport.close_calls == 1
+
+
+class BodyRecordingTransport(NonDuplexTransport):
+    def __init__(self) -> None:
+        super().__init__()
+        self.bodies: list[bytes] = []
+
+    async def send(self, request: Any) -> Any:
+        self.bodies.append(await AsyncBytesReader(request.body).read())
+        return await super().send(request)
+
+
+async def _chunks() -> AsyncIterator[bytes]:
+    yield b"foobar"
+
+
+def _streaming_call_failing_once(
+    harness: Any, body: Any, *, operation: Any = DATA_STREAM_OPERATION
+) -> Any:
+    request = StubRequest()
+    request.body = body
+    harness.protocol.serialize_request = lambda **kwargs: request  # type: ignore[method-assign]
+
+    async def fail_first_attempt(**kwargs: Any) -> StubOutput:
+        if harness.protocol.deserialize_response_calls == 0:
+            harness.protocol.deserialize_response_calls += 1
+            raise CallError("Service unavailable", is_retry_safe=True)
+        return StubOutput()
+
+    harness.protocol.deserialize_response = fail_first_attempt  # type: ignore[method-assign]
+    return replace(
+        client_call(),
+        operation=operation,
+        retry_strategy=SimpleRetryStrategy(
+            backoff_strategy=ExponentialRetryBackoffStrategy(backoff_scale_value=0),
+            max_attempts=2,
+        ),
+    )
+
+
+async def test_seekable_data_stream_is_retried_from_its_start() -> None:
+    harness = pipeline_harness(BodyRecordingTransport())
+    body = BytesIO(b"foobar")
+    body.seek(2)
+    call = _streaming_call_failing_once(harness, AsyncBytesReader(body))
+
+    await harness.pipeline(call)
+
+    # The caller's position is where the body starts, so retries don't resend 'fo'.
+    assert cast(BodyRecordingTransport, harness.transport).bodies == [b"obar", b"obar"]
+
+
+class UnrewindableBody:
+    """A body that reports it's seekable but fails to seek."""
+
+    def seekable(self) -> bool:
+        return True
+
+    def seek(self, offset: int, whence: int = 0) -> int:
+        raise UnsupportedOperation("seek")
+
+    async def read(self, size: int = -1) -> bytes:
+        return b"foobar"
+
+
+@pytest.mark.parametrize(
+    "factory",
+    [lambda: AsyncBytesReader(_chunks()), UnrewindableBody],
+    ids=["not-seekable", "fails-to-seek"],
+)
+async def test_data_stream_that_cant_be_rewound_is_sent_once(
+    factory: Callable[[], Any],
+) -> None:
+    harness = pipeline_harness(BodyRecordingTransport())
+    call = _streaming_call_failing_once(harness, factory())
+
+    with pytest.raises(CallError, match="Service unavailable"):
+        await harness.pipeline(call)
+
+    assert cast(BodyRecordingTransport, harness.transport).bodies == [b"foobar"]
+
+
+class SeekBackFailsBody(UnrewindableBody):
+    """A body that reports its position but fails to seek back to it."""
+
+    def seek(self, offset: int, whence: int = 0) -> int:
+        if whence == 1:
+            return 0
+        raise UnsupportedOperation("seek")
+
+
+async def test_data_stream_that_fails_to_rewind_returns_the_attempts_error() -> None:
+    harness = pipeline_harness(BodyRecordingTransport())
+    call = _streaming_call_failing_once(harness, SeekBackFailsBody())
+
+    with pytest.raises(CallError, match="Service unavailable") as exc_info:
+        await harness.pipeline(call)
+
+    assert cast(BodyRecordingTransport, harness.transport).bodies == [b"foobar"]
+    assert any(
+        note.startswith("Retry not attempted") for note in exc_info.value.__notes__
+    )
+
+
+async def test_input_event_stream_is_not_retried() -> None:
+    harness = pipeline_harness(NonDuplexTransport())
+    call = _streaming_call_failing_once(harness, BytesIO(b""), operation=OPERATION)
+
+    stream = await harness.pipeline.input_stream(call, StubEvent)
+
+    with pytest.raises(CallError, match="Service unavailable"):
+        await stream.await_output()
+    assert harness.transport.send_calls == 1

packages/smithy-http/.changes/next-release/smithy-http-bugfix-dbdfcc08b0ac4033a2f69774df0decdd.json

@@ -0,0 +1,4 @@
+{
+  "type": "bugfix",
+  "description": "Fixed HTTP serializers sending seekable streaming payloads, such as files, without a Content-Length and sending a modeled Content-Length twice."
+}