diff --git a/CHANGELOG.md b/CHANGELOG.md index 24f9c6db..5b931daa 100644 --- a/CHANGELOG.md +++ b/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 `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`, diff --git a/README.md b/README.md index b87e920c..9cbe0305 100644 --- a/README.md +++ b/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 diff --git a/codegen/smithy-cpp-codegen/src/main/java/io/smithycpp/codegen/ClientGenerator.java b/codegen/smithy-cpp-codegen/src/main/java/io/smithycpp/codegen/ClientGenerator.java index 8038d37a..fbaac7b7 100644 --- a/codegen/smithy-cpp-codegen/src/main/java/io/smithycpp/codegen/ClientGenerator.java +++ b/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 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 " + "transport, std::string path_prefix);", name); - w.write( - "opal::Outcome " - + "Send(opal::http::HttpRequest request) const;"); + if (streamsAnyPayload()) { + w.write( + "opal::Outcome " + + "Send(opal::http::HttpRequest request, " + + "const opal::http::BodySink& sink = {}) const;"); + } else { + w.write( + "opal::Outcome " + + "Send(opal::http::HttpRequest request) const;"); + } w.write(""); w.write("opal::ClientConfig config_;"); w.write("std::shared_ptr transport_;"); @@ -400,10 +445,18 @@ private void writeSource(CppWriter w) { w.dedent(); w.write(""); - w.openBlock( - "opal::Outcome $L::Send(" - + "opal::http::HttpRequest request) const {", - name); + boolean sinkOnSend = streamsAnyPayload(); + if (sinkOnSend) { + w.openBlock( + "opal::Outcome $L::Send(" + + "opal::http::HttpRequest request, const opal::http::BodySink& sink) const {", + name); + } else { + w.openBlock( + "opal::Outcome $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;"); } diff --git a/codegen/smithy-cpp-codegen/src/main/java/io/smithycpp/codegen/DirectedCppCodegen.java b/codegen/smithy-cpp-codegen/src/main/java/io/smithycpp/codegen/DirectedCppCodegen.java index e0baac2a..fe230be1 100644 --- a/codegen/smithy-cpp-codegen/src/main/java/io/smithycpp/codegen/DirectedCppCodegen.java +++ b/codegen/smithy-cpp-codegen/src/main/java/io/smithycpp/codegen/DirectedCppCodegen.java @@ -63,6 +63,9 @@ public void generateService(GenerateServiceDirective 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; diff --git a/codegen/smithy-cpp-codegen/src/main/java/io/smithycpp/codegen/HttpBindingCodeGen.java b/codegen/smithy-cpp-codegen/src/main/java/io/smithycpp/codegen/HttpBindingCodeGen.java index 205ffbc6..edb1f86c 100644 --- a/codegen/smithy-cpp-codegen/src/main/java/io/smithycpp/codegen/HttpBindingCodeGen.java +++ b/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 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. + * + *

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. + * + *

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 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. + * + *

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 labels, diff --git a/codegen/smithy-cpp-codegen/src/main/java/io/smithycpp/codegen/HttpJsonBindingProtocol.java b/codegen/smithy-cpp-codegen/src/main/java/io/smithycpp/codegen/HttpJsonBindingProtocol.java index 617ee07e..9689ff25 100644 --- a/codegen/smithy-cpp-codegen/src/main/java/io/smithycpp/codegen/HttpJsonBindingProtocol.java +++ b/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) { diff --git a/codegen/smithy-cpp-codegen/src/main/java/io/smithycpp/codegen/HttpJsonClientGenerator.java b/codegen/smithy-cpp-codegen/src/main/java/io/smithycpp/codegen/HttpJsonClientGenerator.java index c8ee71ce..e041679e 100644 --- a/codegen/smithy-cpp-codegen/src/main/java/io/smithycpp/codegen/HttpJsonClientGenerator.java +++ b/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 diff --git a/codegen/smithy-cpp-codegen/src/main/java/io/smithycpp/codegen/ProtocolGenerator.java b/codegen/smithy-cpp-codegen/src/main/java/io/smithycpp/codegen/ProtocolGenerator.java index 3563d8e6..d908be89 100644 --- a/codegen/smithy-cpp-codegen/src/main/java/io/smithycpp/codegen/ProtocolGenerator.java +++ b/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); diff --git a/codegen/smithy-cpp-codegen/src/test/java/io/smithycpp/codegen/GeneratedCodeShapeTest.java b/codegen/smithy-cpp-codegen/src/test/java/io/smithycpp/codegen/GeneratedCodeShapeTest.java index 7c1accb9..8228a42e 100644 --- a/codegen/smithy-cpp-codegen/src/test/java/io/smithycpp/codegen/GeneratedCodeShapeTest.java +++ b/codegen/smithy-cpp-codegen/src/test/java/io/smithycpp/codegen/GeneratedCodeShapeTest.java @@ -1,9 +1,11 @@ package io.smithycpp.codegen; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import org.junit.jupiter.api.Test; +import software.amazon.smithy.codegen.core.CodegenException; /** * Exactly-once / absence pins for "generator emitted redundant/dead code" fixes with no @@ -395,12 +397,13 @@ void typedErrorListingNameCollisionFailsWithContext() { @Test void streamingBlobsStayPlainBufferedBlobs() { - // The README's "Current limitations": @streaming BLOBS remain unmodeled — - // a streaming blob payload generates as an ordinary, fully buffered - // opal::Blob with the plain unary operation around it. Event-stream - // unions became real in Phase 8 slice 3 (ADR-0016; the flipped pin is - // eventStreamOperationsGenerateStreamingSignatures below), which is why - // this pin is now blob-specific. + // A @streaming blob in the *request* payload is still an ordinary, fully + // buffered opal::Blob with the plain unary operation around it: writing + // one needs chunked request framing, which the http1 codec refuses on + // purpose. Only the response half streams (#213 slice 2, the two tests + // above). Event-stream unions became real in Phase 8 slice 3 (ADR-0016; + // the flipped pin is eventStreamOperationsGenerateStreamingSignatures + // below), which is why this pin is blob-specific. String model = """ $version: "2.0" @@ -433,6 +436,269 @@ void streamingBlobsStayPlainBufferedBlobs() { assertFalse(client.contains("EventStream"), client); } + @Test + void streamingBlobOutputsTakeAWriterAndAreGatedOnSuccess() { + // #213 slice 2. A @streaming blob in the *response* streams to a writer + // the caller supplies rather than materializing in the output. Requests + // keep the pin above: writing one needs chunked request framing, which + // the http1 codec refuses on purpose. + String model = + """ + $version: "2.0" + namespace test.shape + use alloy#simpleRestJson + + @simpleRestJson + service Svc { version: "1", operations: [Download] } + + @readonly + @http(method: "GET", uri: "/download/{id}") + operation Download { + input := { + @required + @httpLabel + id: String + } + output := { + @httpHeader("ETag") + etag: String + + @required + @httpPayload + content: StreamingBlob + } + } + + @streaming + blob StreamingBlob + """; + var manifest = PluginTestHarness.generate(model, "test.shape#Svc", "test::shape"); + + // The writer is defaulted, so an operation that was callable before still + // is, and omitting it buffers exactly as it used to. + String client = manifest.expectFileString("/include/test/shape/client.h"); + assertTrue( + client.contains( + "opal::Outcome Download(const DownloadInput& input, " + + "const opal::http::BodyWriter& write = nullptr) const;"), + client); + + // The accept gate belongs to the generated code, and it is the operation's + // own success condition spelled once more: a modeled code here, so exactly + // that code streams. A wider gate would stream a status the client is + // about to reject, leaving the error path an empty body to parse. + String source = manifest.expectFileString("/src/client.cc"); + assertTrue( + source.contains( + ".accept = [](int status, const opal::http::Headers&) " + "{ return status == 200; },"), + source); + assertTrue(source.contains(".write = write,"), source); + assertTrue(source.contains("auto response = Send(std::move(request), payload_sink);"), source); + // The gate and the status check below it are the same predicate. If they + // ever diverge, either a status the client rejects was streamed (leaving + // the error path an empty body) or one it accepts was buffered. + assertTrue(source.contains("if (response->status != 200) return"), source); + + // A null writer has to reach the buffered path, or omitting the argument + // would turn every such call into an empty-sink streaming send. + assertTrue(source.contains("if (sink.write == nullptr) {"), source); + + // The member stays on the output struct. It is shared with the server + // generator, which still returns the payload; removing it would drag the + // deferred server half into this change. (Smithy requires @required or + // @default on a streaming member, so it is a plain Blob — left empty when + // the bytes went to the writer instead.) + String types = manifest.expectFileString("/include/test/shape/types.h"); + assertTrue(types.contains("opal::Blob content"), types); + } + + @Test + void aStreamingPayloadOnARetryableSuccessStatusIsRefused() { + // #213 slice 2, second cursor finding on PR 216. A model may declare a + // modeled success status that the retry layer classifies as transient — + // @http(code: 503) with the HttpResponseCodeSemantics suppression the + // redirect fixture already uses for 302. The two layers then disagree + // irreconcilably: the generated gate and status check both call 503 a + // success, while SendWithRetries retries it and withholds it from the sink + // on every attempt, so the call returns success with the payload buffered + // into the member and the caller's writer never invoked. + // + // Teaching the retry layer this operation's success predicate is a change + // to a public runtime API and to every client's retry behavior, so this + // slice refuses the model by name instead of honoring it wrongly. The + // generator has no third option: emitting a writer that cannot fire is + // exactly the silent failure this diagnostic exists to prevent. + String model = + """ + $version: "2.0" + namespace test.shape + use alloy#simpleRestJson + + @simpleRestJson + service Svc { version: "1", operations: [Download] } + + @readonly + @suppress(["HttpResponseCodeSemantics"]) + @http(method: "GET", uri: "/download", code: 503) + operation Download { + output := { + @required + @httpPayload + content: StreamingBlob + } + } + + @streaming + blob StreamingBlob + """; + + CodegenException thrown = + assertThrows( + CodegenException.class, + () -> PluginTestHarness.generate(model, "test.shape#Svc", "test::shape")); + String message = thrown.getMessage(); + assertTrue(message.startsWith("cpp-codegen: "), message); + assertTrue(message.contains("test.shape#Download"), message); + assertTrue(message.contains("503"), message); + // The diagnostic has to name the way out, not just the problem. + assertTrue(message.contains("@httpResponseCode"), message); + } + + @Test + void aNonRetryableModeledSuccessStatusStillStreams() { + // The refusal is scoped to the statuses the retry layer treats as + // transient. A modeled 302 — the redirect fixture's own case — is not one + // of them, so it streams on exactly that status. + String model = + """ + $version: "2.0" + namespace test.shape + use alloy#simpleRestJson + + @simpleRestJson + service Svc { version: "1", operations: [Download] } + + @readonly + @suppress(["HttpResponseCodeSemantics"]) + @http(method: "GET", uri: "/download", code: 302) + operation Download { + output := { + @required + @httpPayload + content: StreamingBlob + } + } + + @streaming + blob StreamingBlob + """; + var manifest = PluginTestHarness.generate(model, "test.shape#Svc", "test::shape"); + + String source = manifest.expectFileString("/src/client.cc"); + assertTrue( + source.contains( + ".accept = [](int status, const opal::http::Headers&) { return status == 302; },"), + source); + } + + @Test + void anRpcProtocolLeavesAStreamingBlobResponseBuffered() { + // The protocol gate (#213 slice 2). jsonRpc2 carries every member inside + // one envelope, so there is no standalone response body to hand over + // piecewise — the blob stays a buffered opal::Blob and the operation + // keeps its plain signature. Pinned because the README and + // production-guide both promise it, and because the alternative is a + // writer parameter that silently never fires. + String model = + """ + $version: "2.0" + namespace test.shape + use smithy.cpp.protocols#jsonRpc2 + + @jsonRpc2 + service Svc { version: "1", operations: [Download] } + + operation Download { + input := { + @required + id: String + } + output := { + @required + content: StreamingBlob + } + } + + @streaming + blob StreamingBlob + """; + var manifest = PluginTestHarness.generate(model, "test.shape#Svc", "test::shape"); + + String client = manifest.expectFileString("/include/test/shape/client.h"); + assertTrue( + client.contains( + "opal::Outcome Download(const DownloadInput& input) const;"), + client); + assertFalse(client.contains("BodyWriter"), client); + // The Send helper stays one-argument too: no service here streams, so + // every RPC client is byte-identical to what it was. + assertFalse(client.contains("BodySink"), client); + String source = manifest.expectFileString("/src/client.cc"); + assertFalse(source.contains("payload_sink"), source); + } + + @Test + void aStreamingPayloadUnderHttpResponseCodeStreamsEverySuccessStatus() { + // The other success rule (#213 slice 2, cursor review on PR 216). With + // @httpResponseCode the service picks the status, so success is 2xx or + // 3xx — a modeled redirect that carries a payload is a success, and a gate + // keyed on 2xx would silently buffer it into the member while still + // returning success, which is not what the caller asked for. + String model = + """ + $version: "2.0" + namespace test.shape + use alloy#simpleRestJson + + @simpleRestJson + service Svc { version: "1", operations: [Download] } + + @readonly + @http(method: "GET", uri: "/download/{id}") + operation Download { + input := { + @required + @httpLabel + id: String + } + output := { + @required + @httpResponseCode + status: Integer + + @required + @httpPayload + content: StreamingBlob + } + } + + @streaming + blob StreamingBlob + """; + var manifest = PluginTestHarness.generate(model, "test.shape#Svc", "test::shape"); + + String source = manifest.expectFileString("/src/client.cc"); + assertTrue( + source.contains( + ".accept = [](int status, const opal::http::Headers&) " + + "{ return status >= 200 && status < 400; },"), + source); + // The gate and the check below it are the same predicate; if they ever + // disagree, one of the two states above is unreachable or wrong. + assertTrue( + source.contains("if (response->status < 200 || response->status >= 400) return"), source); + } + @Test void eventStreamOperationsGenerateStreamingSignatures() { // The flip of the old "@streaming is ignored" pin (ADR-0016): an diff --git a/codegen/smithy-cpp-codegen/src/test/java/io/smithycpp/codegen/RetryableStatusMirrorTest.java b/codegen/smithy-cpp-codegen/src/test/java/io/smithycpp/codegen/RetryableStatusMirrorTest.java new file mode 100644 index 00000000..2684c1c9 --- /dev/null +++ b/codegen/smithy-cpp-codegen/src/test/java/io/smithycpp/codegen/RetryableStatusMirrorTest.java @@ -0,0 +1,68 @@ +package io.smithycpp.codegen; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.Set; +import java.util.TreeSet; +import java.util.regex.Matcher; +import java.util.regex.Pattern; +import org.junit.jupiter.api.Test; + +/** + * The generator refuses a @streaming response payload whose modeled success status is one the retry + * layer treats as transient (issue #213 slice 2), which means the Java side carries a second copy + * of a set whose source of truth is C++: {@code opal::RetryableStatus} in + * runtime/src/client/retry.cc. Two copies drift, and the drift would be silent in the direction + * that matters — a status added there but not here goes back to generating a writer the retry layer + * can never invoke. + * + *

So this reads the C++ and compares. The same self-policing style as {@link + * DiagnosticConventionTest}: the mirror is allowed to exist because a test fails when it stops + * being one. + */ +class RetryableStatusMirrorTest { + + private static final Path RETRY_CC = + Paths.get(System.getProperty("smithycpp.repoRoot"), "runtime/src/client/retry.cc"); + + /** RetryableStatus's one-line body. */ + private static final Pattern BODY = + Pattern.compile("bool RetryableStatus\\(int status\\) \\{\\s*return([^}]*)\\}"); + + /** One status literal compared against `status` inside that body. */ + private static final Pattern STATUS = Pattern.compile("status == (\\d{3})"); + + @Test + void theGeneratorsRetryableStatusesAreTheRuntimes() throws IOException { + String text = Files.readString(RETRY_CC); + Matcher matcher = BODY.matcher(text); + assertTrue(matcher.find(), "RetryableStatus(int status) not found in " + RETRY_CC); + + // Compared as the digit tokens both sides are written with, rather than as + // parsed ints: the comparison is exact either way, and there is no parse to + // fail on input the pattern has already constrained to three digits. + Set runtime = new TreeSet<>(); + Matcher statuses = STATUS.matcher(matcher.group(1)); + while (statuses.find()) { + runtime.add(statuses.group(1)); + } + assertFalse(runtime.isEmpty(), "no status literals found in RetryableStatus in " + RETRY_CC); + + Set generator = new TreeSet<>(); + for (Integer status : HttpBindingCodeGen.RETRYABLE_STATUSES) { + generator.add(String.valueOf(status)); + } + + assertEquals( + generator, + runtime, + "HttpBindingCodeGen.RETRYABLE_STATUSES has drifted from opal::RetryableStatus in " + + RETRY_CC); + } +} diff --git a/docs/generated-types.md b/docs/generated-types.md index bb65b084..f2202d08 100644 --- a/docs/generated-types.md +++ b/docs/generated-types.md @@ -22,7 +22,7 @@ compatibility contract: changes to it are breaking for consumers of generated co | `intEnum` | `enum class X : std::int32_t` | Wire values outside int32 fail the parse; unknown in-range values are preserved (servers additionally validate membership) | | `smithy.api#Unit` | `opal::Unit` | Never declared; maps to the runtime type | | `bigInteger` / `bigDecimal` | — | Rejected with a clear error (planned) | -| `@streaming` blob member | trait ignored | Generates as a fully buffered `opal::Blob`; see the README's [Current limitations](../README.md#current-limitations) | +| `@streaming` blob member | `opal::Blob` + a response writer | Always generates as an `opal::Blob` (Smithy requires `@required`/`@default` on it). When it is the *response* `@httpPayload` of an HTTP-binding operation, the client method also takes a defaulted `const opal::http::BodyWriter&`: pass one and the bytes stream to it with the member left empty, omit it and the payload buffers into the member. Request payloads and the RPC protocols stay buffered — see [production-guide.md](production-guide.md#through-a-generated-client) | | `@streaming` union member | typed event stream (ADR-0016) | The operation generates `opal::eventstream::EventStream` signatures (client and server) instead of carrying the union in the body; the union itself still generates as a normal union type | | recursive structures | `opal::Boxed` member indirection | Deep copy/equality; list cycles ride `std::vector` directly. Cycles through union members or map values are still rejected with a clear error | diff --git a/docs/production-guide.md b/docs/production-guide.md index ad5c2ea4..ce60c550 100644 --- a/docs/production-guide.md +++ b/docs/production-guide.md @@ -93,11 +93,70 @@ take the first back. Such a body arrives in `response.body` instead, where error documents already go. `opal::SendWithRetries` has an overload that takes a sink and applies this rule. -Generated clients do not expose a sink yet — a `@streaming` blob member still -generates a fully buffered `opal::Blob`. That is -[#213](https://github.com/muchq/opal-cpp/issues/213) slice 2; today a caller -that needs this drives the transport directly, as -`examples/bazel-consumer/response_sink_acceptance_test.cc` does. +### Through a generated client + +A model that marks the response payload `@streaming` does not need any of the +above. The operation takes an `opal::http::BodyWriter` and the generated code +assembles the sink: + +```smithy +@readonly +@http(method: "GET", uri: "/s/{slug}") +operation Download { + input := { @required @httpLabel slug: String } + output := { + @httpHeader("ETag") etag: String + @required @httpPayload content: StreamingBlob + } +} + +@streaming +blob StreamingBlob +``` + +```cpp +std::ofstream out("export.pgn", std::ios::binary); +auto downloaded = client.Download(DownloadInput{.slug = "big"}, [&](std::string_view piece) { + return out.write(piece.data(), piece.size()).good(); +}); +// downloaded->etag is deserialized as usual; downloaded->content is empty — +// the bytes went to the writer. +``` + +- **The accept gate is the generator's, and it is the operation's own success + condition** — the modeled `@http` code, or `2xx`/`3xx` when the status comes + from `@httpResponseCode`. Success streams the payload; anything else stays + buffered, so a modeled error still deserializes into the typed + `Errors` listing from its own body. A modeled 3xx that carries a + payload is a success, and streams. +- **The writer is defaulted.** `client.Download(input)` with no writer buffers + the payload into the member, exactly as an operation without `@streaming` + does, so adding the trait 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` — left + empty when the bytes went to the writer. The server half still returns it. +- **A writer returning false fails the call, not retryably** — the bytes it + refused are gone, and a retry would only deliver them again. +- **A modeled success status the retry layer retries is refused at generation + time.** `@http(code: 503)` with the `HttpResponseCodeSemantics` suppression + makes 503 this operation's success, but `SendWithRetries` classifies it as + transient and withholds it from the sink on every attempt — the writer could + never fire. The generator names the operation and the fix rather than + emitting a method that cannot keep its contract; bind the status with + `@httpResponseCode` instead. Only a static modeled code can collide: + under `@httpResponseCode` success is 2xx/3xx, which shares nothing with the + retryable set (429, 500, 502, 503, 504). + +Only a response payload streams, and only on the HTTP-binding protocols. +Smithy already forces the `@httpPayload` binding on a streaming blob whenever +the protocol supports it, so there is no in-between case there; on the RPC +protocols, which carry every member in one document, a `@streaming` blob is +base64 inside that document and stays a buffered `opal::Blob`. So does a +request payload — writing one needs chunked request framing, which the http1 +codec refuses on purpose. + +`examples/bazel-consumer/response_sink_acceptance_test.cc` is the out-of-tree +acceptance for both levels. ## Retries diff --git a/docs/server-guide.md b/docs/server-guide.md index d193b493..0960273b 100644 --- a/docs/server-guide.md +++ b/docs/server-guide.md @@ -537,6 +537,7 @@ serialization error) — see [production-guide.md](production-guide.md). Nested `@required` absences as `fieldList` entries and a server-strict serde variant (clients must skip null dense-map values and accept UTC-offset timestamps in responses; servers share -that serde today), and `@streaming` blob payloads (see -[Current limitations](../README.md#current-limitations)). Event-stream operations generate +that serde today), and the server half of `@streaming` blob payloads — a handler still +returns the whole payload in its output, and only the *client* streams it (see +[production-guide.md](production-guide.md#through-a-generated-client)). Event-stream operations generate streaming handlers and a `StreamRouter()` (ADR-0016). diff --git a/examples/bazel-consumer/BUILD.bazel b/examples/bazel-consumer/BUILD.bazel index b99c4841..edac5f59 100644 --- a/examples/bazel-consumer/BUILD.bazel +++ b/examples/bazel-consumer/BUILD.bazel @@ -125,9 +125,9 @@ cc_test( ], ) -# Acceptance for the response body sink (issue #213): a consumer pulling a -# blob payload off a generated server without the body landing in a response -# object, and leaving a modeled error's document where the client looks for it. +# Acceptance for the response body sink (issue #213), both levels: the +# transport-level BodySink a consumer drives itself, and the BodyWriter the +# generator puts on an operation whose payload the model marks @streaming. cc_test( name = "response_sink_acceptance_test", size = "small", @@ -137,6 +137,7 @@ cc_test( ":redirector_client", ":redirector_server", "@googletest//:gtest_main", + "@opal_cpp//runtime:client", "@opal_cpp//runtime:core", "@opal_cpp//runtime:http", "@opal_cpp//runtime:server", diff --git a/examples/bazel-consumer/head_e2e_test.cc b/examples/bazel-consumer/head_e2e_test.cc index 8941f349..6398d278 100644 --- a/examples/bazel-consumer/head_e2e_test.cc +++ b/examples/bazel-consumer/head_e2e_test.cc @@ -31,6 +31,10 @@ namespace { +using acme::redirect::DownloadDynamicInput; +using acme::redirect::DownloadDynamicOutput; +using acme::redirect::DownloadInput; +using acme::redirect::DownloadOutput; using acme::redirect::FetchInput; using acme::redirect::FetchOutput; using acme::redirect::NoSuchSlug; @@ -71,6 +75,21 @@ class ProbeHandler final : public RedirectorHandler { return FetchOutput{.status = 200, .etag = kEtag, .content = opal::Blob::FromString(kContent)}; } + opal::Outcome DownloadDynamic( + const DownloadDynamicInput& input, const opal::server::RequestContext&) override { + if (input.slug != "abc") return NotFound(input.slug); + return DownloadDynamicOutput{ + .status = 200, .etag = kEtag, .content = opal::Blob::FromString(kContent)}; + } + + // The streaming sibling of Fetch (#213). Not this file's subject; it + // answers the same resource so the files cannot drift on what is served. + opal::Outcome Download(const DownloadInput& input, + const opal::server::RequestContext&) override { + if (input.slug != "abc") return NotFound(input.slug); + return DownloadOutput{.etag = kEtag, .content = opal::Blob::FromString(kContent)}; + } + opal::Outcome Resolve(const ResolveInput& input, const opal::server::RequestContext&) override { return NotFound(input.slug); diff --git a/examples/bazel-consumer/model/redirector.smithy b/examples/bazel-consumer/model/redirector.smithy index 5ac42ff4..2205775f 100644 --- a/examples/bazel-consumer/model/redirector.smithy +++ b/examples/bazel-consumer/model/redirector.smithy @@ -12,7 +12,7 @@ namespace acme.redirect /// generator, and only one of them used to work. service Redirector { version: "2026-01-01" - operations: [Resolve, ResolveDynamic, Fetch, Probe] + operations: [Resolve, ResolveDynamic, Fetch, Probe, Download, DownloadDynamic] } /// The status is fixed, so it rides the @http trait. @@ -133,3 +133,71 @@ structure NoSuchSlug { @required message: String } + +/// The case the response sink exists for (issue #213): a payload whose size +/// the model does not bound. @streaming on the blob is the client's cue to +/// hand the bytes to a caller-supplied writer as they arrive rather than +/// build an output structure holding all of them at once. +/// +/// Deliberately a sibling of Fetch rather than a change to it: the two differ +/// only in that trait, so the acceptance test can put the streamed and the +/// buffered spelling of one download side by side. +/// +/// The status is fixed here, so the generated client streams exactly it — +/// DownloadDynamic below is the other branch. +@readonly +@http(method: "GET", uri: "/s/{slug}") +operation Download { + input := { + @required + @httpLabel + slug: String + } + + output := { + @httpHeader("ETag") + etag: String + + @required + @httpPayload + content: StreamingBlob + } + + errors: [NoSuchSlug] +} + +/// Smithy requires @required or @default on a member targeting a streaming +/// blob, so the member is a plain opal::Blob — left empty when the client +/// streamed the bytes to a writer instead of buffering them. +@streaming +blob StreamingBlob + +/// Download with the status chosen per request, the same split Resolve and +/// ResolveDynamic are: a modeled 3xx is a success under @httpResponseCode, so +/// a client gating the payload on 2xx would quietly buffer it into the member +/// and still return success. Both spellings are here because they take +/// different branches through the generator. +@readonly +@http(method: "GET", uri: "/sd/{slug}") +operation DownloadDynamic { + input := { + @required + @httpLabel + slug: String + } + + output := { + @required + @httpResponseCode + status: Integer + + @httpHeader("ETag") + etag: String + + @required + @httpPayload + content: StreamingBlob + } + + errors: [NoSuchSlug] +} diff --git a/examples/bazel-consumer/redirect_e2e_test.cc b/examples/bazel-consumer/redirect_e2e_test.cc index 54483095..ceb393e6 100644 --- a/examples/bazel-consumer/redirect_e2e_test.cc +++ b/examples/bazel-consumer/redirect_e2e_test.cc @@ -29,6 +29,10 @@ namespace { +using acme::redirect::DownloadDynamicInput; +using acme::redirect::DownloadDynamicOutput; +using acme::redirect::DownloadInput; +using acme::redirect::DownloadOutput; using acme::redirect::FetchInput; using acme::redirect::FetchOutput; using acme::redirect::NoSuchSlug; @@ -61,6 +65,23 @@ class SlugHandler final : public RedirectorHandler { targets_["cached"] = "https://example.com/cached"; } + opal::Outcome DownloadDynamic( + const DownloadDynamicInput& input, const opal::server::RequestContext&) override { + auto target = Lookup(input.slug); + if (!target) return std::move(target).error(); + return DownloadDynamicOutput{ + .status = kFound, .etag = kEtag, .content = opal::Blob::FromString(kContent)}; + } + + // The streaming sibling of Fetch (#213). Not this file's subject; it + // answers the same resource so the files cannot drift on what is served. + opal::Outcome Download(const DownloadInput& input, + const opal::server::RequestContext&) override { + auto target = Lookup(input.slug); + if (!target) return std::move(target).error(); + return DownloadOutput{.etag = kEtag, .content = opal::Blob::FromString(kContent)}; + } + opal::Outcome Resolve(const ResolveInput& input, const opal::server::RequestContext&) override { auto target = Lookup(input.slug); diff --git a/examples/bazel-consumer/response_cap_acceptance_test.cc b/examples/bazel-consumer/response_cap_acceptance_test.cc index e8419b13..be789d65 100644 --- a/examples/bazel-consumer/response_cap_acceptance_test.cc +++ b/examples/bazel-consumer/response_cap_acceptance_test.cc @@ -24,6 +24,10 @@ namespace { +using acme::redirect::DownloadDynamicInput; +using acme::redirect::DownloadDynamicOutput; +using acme::redirect::DownloadInput; +using acme::redirect::DownloadOutput; using acme::redirect::FetchInput; using acme::redirect::FetchOutput; using acme::redirect::NoSuchSlug; @@ -53,6 +57,24 @@ class LargePayloadHandler final : public RedirectorHandler { return ProbeOutput{.etag = "\"big\"", .content = opal::Blob::FromString(std::string(kPayloadBytes, 'p'))}; } + opal::Outcome DownloadDynamic( + const DownloadDynamicInput& input, const opal::server::RequestContext&) override { + (void)input; + return DownloadDynamicOutput{ + .status = 200, + .etag = "\"big\"", + .content = opal::Blob::FromString(std::string(kPayloadBytes, 'p'))}; + } + + // The streaming sibling of Fetch (#213). Not this file's subject; it + // answers the same resource so the files cannot drift on what is served. + opal::Outcome Download(const DownloadInput& input, + const opal::server::RequestContext&) override { + (void)input; + return DownloadOutput{.etag = "\"big\"", + .content = opal::Blob::FromString(std::string(kPayloadBytes, 'p'))}; + } + opal::Outcome Resolve(const ResolveInput& input, const opal::server::RequestContext&) override { return NotFound(input.slug); diff --git a/examples/bazel-consumer/response_sink_acceptance_test.cc b/examples/bazel-consumer/response_sink_acceptance_test.cc index e2801a25..762bb70d 100644 --- a/examples/bazel-consumer/response_sink_acceptance_test.cc +++ b/examples/bazel-consumer/response_sink_acceptance_test.cc @@ -1,17 +1,22 @@ // Out-of-tree acceptance for the response body sink (issue #213): a consumer -// module reaching `opal::http::BodySink` through `@opal_cpp//runtime:http` -// and pulling a blob payload off a generated server without the body landing -// in a response object. +// module pulling a blob payload off a generated server without the body +// landing in a response object, at both levels the runtime offers it. // -// This is the shape a consumer has today, and deliberately so: the sink lives -// on the transport, so a caller drives it directly. Slice 2 of #213 is what -// puts a `@streaming` blob behind a generated method; until then this test is -// the record of what the runtime alone gives you. +// The transport level came first: `opal::http::BodySink` through +// `@opal_cpp//runtime:http`, driven by the caller, which is still what an +// operation whose payload the model does not mark `@streaming` gets. Slice 2 +// added the generated level: a `@streaming` blob payload puts an +// `opal::http::BodyWriter` on the operation itself, and the generated code +// owns the sink's accept gate. Both are here because both are supported, and +// because the second is built on the first — a regression in the sink shows +// up in the generated path too. // -// The transport here is `SocketHttpClient`, which inherits the default -// `SendStreaming` — it delivers through the sink but buffers on the way, so -// what this test pins is the contract (the sink gets the bytes, the response -// does not) rather than the memory bound. The bound is `BeastHttpClient`'s, +// The transport underneath is `SocketHttpClient` both times — directly in the +// first half, and as what `RedirectorClient::Create` builds from an endpoint +// in the second. It inherits the default `SendStreaming`, delivering through +// the sink but buffering on the way, so what this file pins is the contract +// (the sink gets the bytes, the response does not) rather than the memory +// bound. The bound is `BeastHttpClient`'s, // pinned in the runtime's own beast_client_test.cc, and a consumer gets it by // injecting that transport instead. Keeping Beast out of this target is what // lets it run behind a download-blocking proxy alongside the other @@ -19,6 +24,7 @@ #include +#include #include #include #include @@ -26,6 +32,7 @@ #include "acme/redirect/client.h" #include "acme/redirect/server.h" +#include "opal/client/config.h" #include "opal/core/error.h" #include "opal/http/message.h" #include "opal/http/socket_transport.h" @@ -33,11 +40,17 @@ namespace { +using acme::redirect::DownloadDynamicInput; +using acme::redirect::DownloadDynamicOutput; +using acme::redirect::DownloadErrors; +using acme::redirect::DownloadInput; +using acme::redirect::DownloadOutput; using acme::redirect::FetchInput; using acme::redirect::FetchOutput; using acme::redirect::NoSuchSlug; using acme::redirect::ProbeInput; using acme::redirect::ProbeOutput; +using acme::redirect::RedirectorClient; using acme::redirect::RedirectorHandler; using acme::redirect::RedirectorServer; using acme::redirect::ResolveDynamicInput; @@ -57,6 +70,14 @@ std::string LargePayload() { return payload; } +// A cheap order-sensitive fingerprint: a test that only counted bytes would +// pass on pieces delivered out of order or duplicated. +std::size_t Digest(std::string_view bytes) { + std::size_t digest = 0; + for (const char byte : bytes) digest = digest * 31 + static_cast(byte); + return digest; +} + class DownloadHandler final : public RedirectorHandler { public: explicit DownloadHandler(std::string payload) : payload_(std::move(payload)) {} @@ -71,6 +92,30 @@ class DownloadHandler final : public RedirectorHandler { const opal::server::RequestContext&) override { return ProbeOutput{.etag = "\"big\"", .content = opal::Blob::FromString(payload_)}; } + // Fetch's @streaming twin: the same bytes under the same rules, so the two + // tests below compare the streamed and the buffered delivery of one + // download rather than of two different ones. The server side is unchanged + // by @streaming — the handler still returns the payload in the output. + opal::Outcome Download(const DownloadInput& input, + const opal::server::RequestContext&) override { + if (input.slug != "big") return NotFound(input.slug); + return DownloadOutput{.etag = "\"big\"", .content = opal::Blob::FromString(payload_)}; + } + // The @httpResponseCode spelling: "moved" answers a modeled 302 that still + // carries the payload, which is a success the client must stream rather + // than quietly buffer. + opal::Outcome DownloadDynamic( + const DownloadDynamicInput& input, const opal::server::RequestContext&) override { + if (input.slug == "big") { + return DownloadDynamicOutput{ + .status = 200, .etag = "\"big\"", .content = opal::Blob::FromString(payload_)}; + } + if (input.slug == "moved") { + return DownloadDynamicOutput{ + .status = 302, .etag = "\"big\"", .content = opal::Blob::FromString(payload_)}; + } + return NotFound(input.slug); + } opal::Outcome Resolve(const ResolveInput& input, const opal::server::RequestContext&) override { return NotFound(input.slug); @@ -132,9 +177,7 @@ TEST_F(ResponseSinkAcceptanceTest, ABlobPayloadArrivesThroughTheSinkAndNotInTheR EXPECT_EQ(received, payload_.size()); EXPECT_TRUE(response->body.empty()) << "the payload was delivered twice"; - std::size_t expected = 0; - for (const char byte : payload_) expected = expected * 31 + static_cast(byte); - EXPECT_EQ(digest, expected) << "the bytes arrived, but not these bytes"; + EXPECT_EQ(digest, Digest(payload_)) << "the bytes arrived, but not these bytes"; } TEST_F(ResponseSinkAcceptanceTest, AModeledErrorIsLeftWhereTheClientLooksForIt) { @@ -163,4 +206,159 @@ TEST_F(ResponseSinkAcceptanceTest, AModeledErrorIsLeftWhereTheClientLooksForIt) EXPECT_NE(response->body.find("no slug: missing"), std::string::npos) << response->body; } +// The generated level (slice 2). The client is built from an endpoint alone — +// no transport injected, no sink assembled by hand — which is the whole point: +// a @streaming blob payload is streamed by the operation the generator wrote. +class StreamingPayloadAcceptanceTest : public ::testing::Test { + protected: + void SetUp() override { + ASSERT_TRUE(transport_.Start(server_.Handler()).ok()); + config_.endpoint = "http://127.0.0.1:" + std::to_string(transport_.port()); + } + void TearDown() override { transport_.Stop(); } + + RedirectorClient Client() { + auto client = RedirectorClient::Create(config_); + EXPECT_TRUE(client.ok()) << client.error().message(); + return *std::move(client); + } + + std::string payload_ = LargePayload(); + RedirectorServer server_{std::make_shared(payload_)}; + opal::http::SocketHttpServer transport_; + opal::ClientConfig config_; +}; + +TEST_F(StreamingPayloadAcceptanceTest, ThePayloadGoesToTheWriterAndTheMemberStaysEmpty) { + RedirectorClient client = Client(); + + std::size_t received = 0; + std::size_t digest = 0; + const auto downloaded = + client.Download(DownloadInput{.slug = "big"}, [&](std::string_view piece) { + received += piece.size(); + for (const char byte : piece) digest = digest * 31 + static_cast(byte); + return true; + }); + + ASSERT_TRUE(downloaded.ok()) << downloaded.error().message(); + EXPECT_EQ(received, payload_.size()); + EXPECT_EQ(digest, Digest(payload_)) << "the bytes arrived, but not these bytes"; + // The other bindings still deserialize — streaming the payload does not cost + // the caller the rest of the output. + EXPECT_EQ(downloaded->etag.value_or(""), "\"big\""); + // Smithy requires @required on a streaming member, so `content` is a plain + // Blob rather than an optional: streamed means empty, not absent. + EXPECT_TRUE(downloaded->content.empty()) << "the payload was delivered twice"; +} + +TEST_F(StreamingPayloadAcceptanceTest, OmittingTheWriterBuffersThePayloadIntoTheMember) { + // The writer is defaulted so the operation stays callable the ordinary way, + // and a caller that does not want to stream is not forced to. + RedirectorClient client = Client(); + + const auto downloaded = client.Download(DownloadInput{.slug = "big"}); + ASSERT_TRUE(downloaded.ok()) << downloaded.error().message(); + EXPECT_EQ(downloaded->content.size(), payload_.size()); + EXPECT_EQ(Digest(downloaded->content.ToString()), Digest(payload_)); +} + +TEST_F(StreamingPayloadAcceptanceTest, AModeledErrorDeserializesInsteadOfReachingTheWriter) { + // The accept gate the generator emits, from the outside: an error document + // is not a payload, so it stays buffered where the typed-error path reads + // it. Without the gate this call would return a 404-shaped success with an + // error document in the caller's writer. + RedirectorClient client = Client(); + + bool wrote = false; + const auto downloaded = client.Download(DownloadInput{.slug = "missing"}, [&](std::string_view) { + wrote = true; + return true; + }); + + ASSERT_FALSE(downloaded.ok()); + EXPECT_FALSE(wrote); + const DownloadErrors modeled = DownloadErrors::FromError(downloaded.error()); + ASSERT_TRUE(modeled.is_no_such_slug()) << downloaded.error().message(); + EXPECT_EQ(modeled.as_no_such_slug().message, "no slug: missing"); +} + +TEST_F(StreamingPayloadAcceptanceTest, AWriterThatAbortsFailsTheCallWithoutRetrying) { + // The caller hit its own limit (a disk quota, a length it refuses to + // exceed). Retrying would deliver the same bytes again, so the failure is + // not retryable and the backoff never runs. + int sleeps = 0; + config_.retry.sleep = [&](std::chrono::milliseconds) { ++sleeps; }; + RedirectorClient client = Client(); + + std::size_t received = 0; + const auto downloaded = + client.Download(DownloadInput{.slug = "big"}, [&](std::string_view piece) { + received += piece.size(); + return received <= payload_.size() / 2; + }); + + ASSERT_FALSE(downloaded.ok()); + EXPECT_FALSE(downloaded.error().retryable()); + EXPECT_EQ(sleeps, 0) << "the retry loop backed off for a body the caller refused"; +} + +TEST_F(StreamingPayloadAcceptanceTest, AModeledRedirectCarryingThePayloadStillStreamsIt) { + // Under @httpResponseCode the service picks the status and 3xx is a success + // the client returns — so the gate has to be the same predicate, not "2xx". + // With a 2xx gate this call still succeeds, which is what makes the bug + // quiet: the payload lands in the member and the writer is never called. + RedirectorClient client = Client(); + + std::size_t received = 0; + std::size_t digest = 0; + const auto downloaded = + client.DownloadDynamic(DownloadDynamicInput{.slug = "moved"}, [&](std::string_view piece) { + received += piece.size(); + for (const char byte : piece) digest = digest * 31 + static_cast(byte); + return true; + }); + + ASSERT_TRUE(downloaded.ok()) << downloaded.error().message(); + EXPECT_EQ(downloaded->status, 302); + EXPECT_EQ(received, payload_.size()); + EXPECT_EQ(digest, Digest(payload_)); + EXPECT_TRUE(downloaded->content.empty()) << "a modeled 3xx payload was buffered, not streamed"; +} + +TEST_F(StreamingPayloadAcceptanceTest, TheDynamicStatusSpellingStreamsAnOrdinary200Too) { + RedirectorClient client = Client(); + + std::size_t received = 0; + const auto downloaded = + client.DownloadDynamic(DownloadDynamicInput{.slug = "big"}, [&](std::string_view piece) { + received += piece.size(); + return true; + }); + + ASSERT_TRUE(downloaded.ok()) << downloaded.error().message(); + EXPECT_EQ(downloaded->status, 200); + EXPECT_EQ(received, payload_.size()); + EXPECT_TRUE(downloaded->content.empty()); +} + +TEST_F(StreamingPayloadAcceptanceTest, TheDynamicSpellingLeavesAModeledErrorBuffered) { + // The failure half of the same predicate: 404 is outside it, so the body + // stays where the typed-error path reads it. + RedirectorClient client = Client(); + + bool wrote = false; + const auto downloaded = + client.DownloadDynamic(DownloadDynamicInput{.slug = "missing"}, [&](std::string_view) { + wrote = true; + return true; + }); + + ASSERT_FALSE(downloaded.ok()); + EXPECT_FALSE(wrote); + EXPECT_TRUE( + acme::redirect::DownloadDynamicErrors::FromError(downloaded.error()).is_no_such_slug()) + << downloaded.error().message(); +} + } // namespace diff --git a/runtime/include/opal/http/transport.h b/runtime/include/opal/http/transport.h index c931e98e..86e7b45b 100644 --- a/runtime/include/opal/http/transport.h +++ b/runtime/include/opal/http/transport.h @@ -25,6 +25,12 @@ struct TlsOptions { std::string ca_pem{}; }; +// The write half of a BodySink, named because callers pass one on its own: +// a generated operation with a @streaming blob response takes a BodyWriter and +// builds the sink around it (issue #213), so the caller never spells out an +// accept() the protocol already decides. +using BodyWriter = std::function; + // Where a response body goes when the caller does not want it buffered // (issue #213). Send() holds a whole body in memory before anything decodes // it, which is the right default for a modeled JSON response and the wrong @@ -49,7 +55,7 @@ struct BodySink { // Called with each piece as it arrives, in order, never empty. The view is // valid only for the duration of the call. False aborts the transfer: the // send fails and the connection is dropped rather than reused. - std::function write; + BodyWriter write; }; // Client-side transport. Implementations: SocketHttpClient (built-in HTTP/1.1