Stream @streaming blob response payloads from generated clients (#213 slice 2)

CHANGELOG.md

@@ -95,6 +95,32 @@ policy in [docs/versioning.md](docs/versioning.md).
   `max_backoff` because that one bounds a guess this client made rather than
   a number it was sent. `opal::RetryAfterDelay` is public for callers driving
   their own retry loop.
+- **Generated clients stream a `@streaming` blob response payload** (#213,
+  slice 2). An operation whose response `@httpPayload` targets a `@streaming`
+  blob takes an `opal::http::BodyWriter` alongside its input: the bytes go to
+  the writer as they arrive and the member is left empty. The generated code
+  owns the sink's accept gate and keys it on the operation's own success
+  condition — the modeled `@http` code, or 2xx/3xx under `@httpResponseCode` —
+  so the streamed and buffered states line up with the success and failure
+  ones: a modeled error still deserializes into the typed `<Operation>Errors`
+  listing from its own body rather than arriving in the caller's writer as an
+  unexplained payload, and a modeled 3xx that carries a payload streams rather
+  than quietly buffering.
+  The writer is defaulted, so `client.Download(input)` still buffers into the
+  member and adding `@streaming` to a model breaks no caller. The member stays
+  on the output structure — Smithy requires `@required` or `@default` on a
+  streaming member, so it is a plain `opal::Blob`, and the server half (which
+  still returns the whole payload) is unchanged. Only a response payload on an
+  HTTP-binding protocol streams — Smithy already forces the `@httpPayload`
+  binding there, so there is no in-between case — while on the RPC protocols,
+  which carry every member in one document, and in a request payload, the blob
+  stays a buffered `opal::Blob` exactly as before. A model whose streaming
+  payload rides a modeled success status the retry layer treats as transient
+  (`@http(code: 503)` and the `HttpResponseCodeSemantics` suppression) fails
+  generation with a diagnostic naming the operation and the fix: the retry
+  loop withholds such a status from the sink on every attempt, so a writer
+  emitted for it could never fire.
+
 - **A response body sink: `HttpClient::SendStreaming`** (#213, slice 1 of
   `@streaming` blob support). A caller that does not want a response body
   buffered passes an `opal::http::BodySink` — an `accept(status, headers)`
@@ -111,8 +137,7 @@ policy in [docs/versioning.md](docs/versioning.md).
   gains an overload taking a sink, and withholds retryable statuses from it
   while retries are enabled — streaming a 503's error document and then
   retrying would hand the sink two bodies' worth of bytes with no way to take
-  the first back. Generated clients do not expose a sink yet; a `@streaming`
-  blob member still generates a buffered `opal::Blob` (#213 slice 2).
+  the first back.
 - **A dependency-free Prometheus `/metrics` endpoint** (#91, first work
   item). `opal::server::MetricsRegistry` aggregates the existing `Observe`
   hooks into the five `http_server_*` families labeled by `service_name`,

README.md

@@ -37,8 +37,14 @@ experience assumed. Day 2 (evolving the model) is
 
 Consolidated in one place — if your API depends on any of these, check here before adopting:
 
-- **`@streaming` blobs are not modeled yet.** A streaming blob payload generates as an
-  ordinary `opal::Blob`, fully buffered in memory. Event streams, by contrast, are real
+- **`@streaming` blobs stream on the client only.** A `@streaming` blob bound as an
+  operation's *response* `@httpPayload` puts a defaulted `opal::http::BodyWriter` on the
+  generated client method: the bytes go to the writer as they arrive and the member is
+  left empty ([production-guide](docs/production-guide.md#through-a-generated-client)).
+  Everywhere else the blob is still an ordinary `opal::Blob`, fully buffered in memory:
+  request payloads (writing one needs chunked request framing), the RPC protocols (every
+  member rides one document), and the whole server half. Event streams, by contrast, are
+  real
   ([ADR-0016](docs/adr/0016-generated-event-streams.md)): a `@streaming` union operation
   generates a typed `opal::eventstream::EventStream` session over WebSocket for all
   three protocols — `simpleRestJson` and `rpcv2Cbor` ride the event-stream framing codec

codegen/smithy-cpp-codegen/src/main/java/io/smithycpp/codegen/ClientGenerator.java

@@ -1,6 +1,7 @@
 package io.smithycpp.codegen;
 
 import java.util.List;
+import software.amazon.smithy.model.shapes.MemberShape;
 import software.amazon.smithy.model.shapes.OperationShape;
 import software.amazon.smithy.model.shapes.ServiceShape;
 import software.amazon.smithy.model.shapes.StructureShape;
@@ -28,6 +29,22 @@ private List<OperationShape> streamingOperations() {
     return EventStreamCodeGen.streamingOperations(context.model(), operations());
   }
 
+  /**
+   * The operation's @streaming blob response payload, or null (issue #213): the member whose bytes
+   * the caller's writer takes instead of the output structure. Protocol-gated — the RPC protocols
+   * have no standalone response body to hand over.
+   */
+  private MemberShape streamedPayload(OperationShape operation) {
+    return protocol.supportsStreamingBlobPayloads()
+        ? HttpBindingCodeGen.streamingResponsePayload(context, operation)
+        : null;
+  }
+
+  /** Whether any operation streams its response payload — what puts a sink on the Send helper. */
+  private boolean streamsAnyPayload() {
+    return operations().stream().anyMatch(op -> streamedPayload(op) != null);
+  }
+
   private String clientName() {
     return CppReservedWords.escape(service.getId().getName()) + "Client";
   }
@@ -116,12 +133,33 @@ private void writeHeader(CppWriter w) {
             defaulted);
         continue;
       }
-      w.write(
-          "opal::Outcome<$L> $L(const $L& input$L) const;",
-          outputType,
-          CppReservedWords.escape(operation.getId().getName()),
-          inputType,
-          defaulted);
+      MemberShape streamed = streamedPayload(operation);
+      if (streamed != null) {
+        if (documented) {
+          w.write("///"); // blank separator: model docs above, boilerplate below
+        }
+        w.write(
+            "/// The @streaming response payload '$L' is handed to `write` in",
+            streamed.getMemberName());
+        w.write("/// pieces as it arrives rather than buffered (issue #213), and the");
+        w.write("/// member is left empty. `write` returning false aborts the transfer");
+        w.write("/// and fails the call. Omitting it buffers the payload into the");
+        w.write("/// member as every other operation does.");
+        w.write(
+            "opal::Outcome<$L> $L(const $L& input$L, "
+                + "const opal::http::BodyWriter& write = nullptr) const;",
+            outputType,
+            CppReservedWords.escape(operation.getId().getName()),
+            inputType,
+            defaulted);
+      } else {
+        w.write(
+            "opal::Outcome<$L> $L(const $L& input$L) const;",
+            outputType,
+            CppReservedWords.escape(operation.getId().getName()),
+            inputType,
+            defaulted);
+      }
       if (pagination(operation).isPresent()) {
         w.write(
             "/// Pages $L until the service stops returning a next token (@paginated).",
@@ -140,9 +178,16 @@ private void writeHeader(CppWriter w) {
         "$L(opal::ClientConfig config, std::shared_ptr<opal::http::HttpClient> "
             + "transport, std::string path_prefix);",
         name);
-    w.write(
-        "opal::Outcome<opal::http::HttpResponse> "
-            + "Send(opal::http::HttpRequest request) const;");
+    if (streamsAnyPayload()) {
+      w.write(
+          "opal::Outcome<opal::http::HttpResponse> "
+              + "Send(opal::http::HttpRequest request, "
+              + "const opal::http::BodySink& sink = {}) const;");
+    } else {
+      w.write(
+          "opal::Outcome<opal::http::HttpResponse> "
+              + "Send(opal::http::HttpRequest request) const;");
+    }
     w.write("");
     w.write("opal::ClientConfig config_;");
     w.write("std::shared_ptr<opal::http::HttpClient> transport_;");
@@ -400,10 +445,18 @@ private void writeSource(CppWriter w) {
     w.dedent();
     w.write("");
 
-    w.openBlock(
-        "opal::Outcome<opal::http::HttpResponse> $L::Send("
-            + "opal::http::HttpRequest request) const {",
-        name);
+    boolean sinkOnSend = streamsAnyPayload();
+    if (sinkOnSend) {
+      w.openBlock(
+          "opal::Outcome<opal::http::HttpResponse> $L::Send("
+              + "opal::http::HttpRequest request, const opal::http::BodySink& sink) const {",
+          name);
+    } else {
+      w.openBlock(
+          "opal::Outcome<opal::http::HttpResponse> $L::Send("
+              + "opal::http::HttpRequest request) const {",
+          name);
+    }
     w.write("// Operations with a non-document response payload set their own accept.");
     w.write(
         "if (!request.headers.Get(\"accept\").has_value()) "
@@ -414,9 +467,22 @@ private void writeSource(CppWriter w) {
     w.openBlock("if (!request.body.empty()) {");
     w.write("request.headers.Set(\"content-length\", std::to_string(request.body.size()));");
     w.closeBlock("}");
-    w.write(
-        "return opal::SendWithRetries(*transport_, request, config_.retry, "
-            + "config_.interceptors);");
+    if (sinkOnSend) {
+      w.write("// A sink with no writer is the caller declining to stream (#213):");
+      w.write("// the response buffers on the same path every other operation takes.");
+      w.openBlock("if (sink.write == nullptr) {");
+      w.write(
+          "return opal::SendWithRetries(*transport_, request, config_.retry, "
+              + "config_.interceptors);");
+      w.closeBlock("}");
+      w.write(
+          "return opal::SendWithRetries(*transport_, request, config_.retry, "
+              + "config_.interceptors, sink);");
+    } else {
+      w.write(
+          "return opal::SendWithRetries(*transport_, request, config_.retry, "
+              + "config_.interceptors);");
+    }
     w.closeBlock("}");
     w.write("");
 
@@ -431,12 +497,22 @@ private void writeSource(CppWriter w) {
                   .cppSymbols()
                   .toSymbol(ProtocolSupport.outputShape(context, operation))
                   .getName();
-      w.openBlock(
-          "opal::Outcome<$L> $L::$L(const $L& input) const {",
-          outputType,
-          name,
-          CppReservedWords.escape(operation.getId().getName()),
-          inputType);
+      if (streamedPayload(operation) != null) {
+        w.openBlock(
+            "opal::Outcome<$L> $L::$L(const $L& input, "
+                + "const opal::http::BodyWriter& write) const {",
+            outputType,
+            name,
+            CppReservedWords.escape(operation.getId().getName()),
+            inputType);
+      } else {
+        w.openBlock(
+            "opal::Outcome<$L> $L::$L(const $L& input) const {",
+            outputType,
+            name,
+            CppReservedWords.escape(operation.getId().getName()),
+            inputType);
+      }
       if (input.members().isEmpty()) {
         w.write("(void)input;");
       }

codegen/smithy-cpp-codegen/src/main/java/io/smithycpp/codegen/DirectedCppCodegen.java

@@ -63,6 +63,9 @@ public void generateService(GenerateServiceDirective<CppContext, CppSettings> di
         // Event-stream scope checks (ADR-0016) fail generation with a named
         // diagnostic before any streaming code is emitted.
         EventStreamCodeGen.validate(directive.context(), service, protocol, operations);
+        // The streaming-payload scope check (#213 slice 2), same posture: a
+        // named diagnostic before any client is emitted.
+        HttpBindingCodeGen.validateStreamingPayloads(directive.context(), protocol, operations);
         if (directive.settings().generateClient()) {
           clientGenerator.run();
           hasClient = true;

codegen/smithy-cpp-codegen/src/main/java/io/smithycpp/codegen/HttpBindingCodeGen.java

@@ -3,6 +3,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.TreeMap;
 import software.amazon.smithy.codegen.core.CodegenException;
 import software.amazon.smithy.model.knowledge.HttpBinding;
@@ -11,8 +12,10 @@
 import software.amazon.smithy.model.shapes.OperationShape;
 import software.amazon.smithy.model.shapes.Shape;
 import software.amazon.smithy.model.shapes.ShapeType;
+import software.amazon.smithy.model.traits.HttpTrait;
 import software.amazon.smithy.model.traits.JsonNameTrait;
 import software.amazon.smithy.model.traits.MediaTypeTrait;
+import software.amazon.smithy.model.traits.StreamingTrait;
 
 /**
  * HTTP-binding emission shared by the client and server halves of the HTTP+JSON protocol (companion
@@ -41,6 +44,84 @@ static String wireName(MemberShape member, boolean useJsonName) {
         .orElse(member.getMemberName());
   }
 
+  /**
+   * The HTTP statuses the retry layer treats as transient. A mirror of {@code
+   * opal::RetryableStatus} (runtime/src/client/retry.cc), which is the source of truth — the
+   * generator cannot call into C++, and {@link RetryableStatusMirrorTest} fails when the two drift.
+   * Used only by {@link #validateStreamingPayloads}: a modeled success status in this set is a
+   * model the streaming payload cannot be honored on.
+   */
+  static final Set<Integer> RETRYABLE_STATUSES = Set.of(429, 500, 502, 503, 504);
+
+  /**
+   * Fails generation for a @streaming response payload whose modeled success status is one the
+   * retry layer retries (issue #213 slice 2). The two layers cannot both be obeyed: the generated
+   * code calls that status a success and streams on it, while {@code SendWithRetries} classifies it
+   * as transient, retries it, and withholds it from the sink on every attempt — so the call would
+   * return success with the payload buffered into the member and the caller's writer never invoked.
+   * Refusing by name beats emitting a writer that cannot fire.
+   *
+   * <p>Only the static-code arm can collide. Under @httpResponseCode success is 2xx/3xx, which
+   * shares nothing with the retryable set, and the RPC protocols do not stream at all.
+   *
+   * <p>The way out is in the diagnostic: @httpResponseCode carries the status at runtime, so the
+   * operation stops modeling a transient code as its success.
+   */
+  static void validateStreamingPayloads(
+      CppContext context, ProtocolGenerator protocol, List<OperationShape> operations) {
+    if (!protocol.supportsStreamingBlobPayloads()) {
+      return;
+    }
+    HttpBindingIndex index = HttpBindingIndex.of(context.model());
+    for (OperationShape operation : operations) {
+      MemberShape streamed = streamingResponsePayload(context, operation);
+      if (streamed == null || ResponseBindings.of(index, operation).responseCode() != null) {
+        continue;
+      }
+      int code = operation.expectTrait(HttpTrait.class).getCode();
+      if (!RETRYABLE_STATUSES.contains(code)) {
+        continue;
+      }
+      throw new CodegenException(
+          "cpp-codegen: operation "
+              + operation.getId()
+              + " streams its @streaming response payload '"
+              + streamed.getMemberName()
+              + "' on modeled status "
+              + code
+              + ", which the retry layer treats as transient (opal::RetryableStatus): it would be"
+              + " retried and withheld from the body sink on every attempt, so the writer could"
+              + " never be invoked. Bind the status with @httpResponseCode so it is chosen at"
+              + " runtime rather than modeled as this operation's success, or drop @streaming"
+              + " from the payload");
+    }
+  }
+
+  /**
+   * The operation's response @httpPayload member when it targets a @streaming blob, else null
+   * (issue #213 slice 2). Such a member is the whole response body and the model puts no bound on
+   * its size, so the generated operation takes an {@code opal::http::BodyWriter} and hands the
+   * bytes to it instead of materializing the member.
+   *
+   * <p>Only an @httpPayload blob qualifies. A @streaming blob bound anywhere else is base64 inside
+   * a JSON document — the document has to be parsed whole before the member exists, so there is
+   * nothing to stream — and it stays a buffered {@code opal::Blob}, exactly as it was.
+   */
+  static MemberShape streamingResponsePayload(CppContext context, OperationShape operation) {
+    if (!operation.hasTrait(HttpTrait.class)) {
+      return null;
+    }
+    HttpBinding payload =
+        ResponseBindings.of(HttpBindingIndex.of(context.model()), operation).payload();
+    if (payload == null) {
+      return null;
+    }
+    Shape target = context.model().expectShape(payload.getMember().getTarget());
+    return target.isBlobShape() && target.hasTrait(StreamingTrait.class)
+        ? payload.getMember()
+        : null;
+  }
+
   /** An operation's request bindings partitioned by location (maps sorted by location name). */
   record RequestBindings(
       Map<String, HttpBinding> labels,

codegen/smithy-cpp-codegen/src/main/java/io/smithycpp/codegen/HttpJsonBindingProtocol.java

@@ -93,6 +93,11 @@ public void writeErrorDocPatches(CppWriter w, CppContext context, StructureShape
     client.writeErrorDocPatches(w, context, error);
   }
 
+  @Override
+  public boolean supportsStreamingBlobPayloads() {
+    return true;
+  }
+
   @Override
   public void writeOperationBody(
       CppWriter w, CppContext context, ServiceShape service, OperationShape operation) {

codegen/smithy-cpp-codegen/src/main/java/io/smithycpp/codegen/HttpJsonClientGenerator.java

@@ -178,7 +178,26 @@ void writeOperationBody(
           HttpBindingCodeGen.payloadContentType(context, operation, false));
     }
     ProtocolSupport.writeRequestCompression(w, operation);
-    w.write("auto response = Send(std::move(request));");
+    if (HttpBindingCodeGen.streamingResponsePayload(context, operation) != null) {
+      // #213: the @streaming payload goes straight to the caller's writer. The
+      // gate is this operation's success condition and nothing else — the same
+      // predicate the status check below applies, so the two states line up:
+      // a success streamed the payload, and a failure left its body buffered
+      // for the error path to parse. A wider gate would stream a status the
+      // client then rejects, leaving that path an empty body; a narrower one
+      // would silently buffer a payload into the member on a status the client
+      // calls success (a modeled 3xx under @httpResponseCode). A null writer
+      // makes this an incomplete sink, which Send() reads as "buffer it".
+      w.openBlock("const opal::http::BodySink payload_sink{");
+      w.write(
+          ".accept = [](int status, const opal::http::Headers&) { return $L; },",
+          responseCode != null ? "status >= 200 && status < 400" : "status == " + http.getCode());
+      w.write(".write = write,");
+      w.closeBlock("};");
+      w.write("auto response = Send(std::move(request), payload_sink);");
+    } else {
+      w.write("auto response = Send(std::move(request));");
+    }
     w.write("if (!response) return std::move(response).error();");
     if (responseCode != null) {
       // The service chooses the status at runtime via @httpResponseCode, so

codegen/smithy-cpp-codegen/src/main/java/io/smithycpp/codegen/ProtocolGenerator.java

@@ -59,6 +59,16 @@ default boolean errorStatusFallback() {
   default void writeErrorDocPatches(
       CppWriter w, CppContext context, software.amazon.smithy.model.shapes.StructureShape error) {}
 
+  /**
+   * Whether a @streaming blob bound as the response @httpPayload streams to a caller-supplied
+   * writer (issue #213): true for the HTTP binding protocols, where the payload is the whole
+   * response body. The RPC protocols put every member inside one document, so there is no body to
+   * hand over piecewise and their @streaming blobs stay buffered, as they always were.
+   */
+  default boolean supportsStreamingBlobPayloads() {
+    return false;
+  }
+
   /** Emits the body of one operation method (inside the function braces). */
   void writeOperationBody(
       CppWriter w, CppContext context, ServiceShape service, OperationShape operation);