Merge pull request #214 from muchq/claude/kind-fermi-elfabg-213-sink

Stream a response body to a sink instead of buffering it (#213 slice 1)

Author: aaylward

CHANGELOG.md

@@ -82,6 +82,24 @@ policy in [docs/versioning.md](docs/versioning.md).
 
 ### Added
 
+- **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)`
+  asked once per response, and a `write(piece)` given the body in order as it
+  arrives — and gets back the status and headers with an empty body. It is
+  what `max_response_bytes` cannot be: that cap bounds what this process
+  holds, and a download whose size the service does not bound needs the
+  process not to hold it. `BeastHttpClient` streams for real and never
+  assembles an accepted body; every other transport inherits a default that
+  buffers and then hands the body over in one piece, so the contract holds
+  everywhere and the memory bound follows the transport. Deciding per response
+  is deliberate: a sink that takes 200s leaves a modeled error's document in
+  `response.body` where the generated client already reads it. `SendWithRetries`
+  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).
 - **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`,

docs/production-guide.md

@@ -53,9 +53,51 @@ to accept 64 MiB.
 config.max_response_bytes = std::size_t{4} * 1024 * 1024;  // 4 MiB: this API pages
 ```
 
-Bodies are still buffered, not streamed; a sink-based transport API is the
-follow-up if a consumer needs bounded memory for large downloads
-(`docs/research/client-third-party-api-gaps.md`, §6).
+## Streaming a response body
+
+The cap above bounds a buffered body. A download whose size the service does
+not bound needs the other thing: not holding it at all. `SendStreaming` takes
+an `opal::http::BodySink` and hands the body over in pieces as it arrives.
+
+```cpp
+std::ofstream out("export.pgn", std::ios::binary);
+const opal::http::BodySink to_file{
+    .accept = [](int status, const opal::http::Headers&) { return status == 200; },
+    .write = [&](std::string_view piece) { return out.write(piece.data(), piece.size()).good(); },
+};
+auto response = transport->SendStreaming(request, to_file);
+```
+
+- **`accept` is asked once per response**, after the status and headers and
+  before any body byte. True streams the body; false buffers it into
+  `response.body` exactly as `Send` would. Deciding per response is what lets
+  a caller take the payload and leave a 404's error document where the rest
+  of the client already knows to read it.
+- **`write` gets each piece in order**, never empty, valid only for the call.
+  Returning false aborts the transfer: the send fails with a non-retryable
+  error and the connection is dropped rather than reused.
+- **On a streamed response `max_response_bytes` does not apply.** The cap
+  bounds what this process holds, and a sink holds nothing here. A declined
+  response is buffered and capped as always.
+- **`BeastHttpClient` is the transport that actually streams.** Every other
+  transport inherits a default that buffers and then hands the body over in
+  one piece: same delivery, same empty `response.body`, no memory bound. So
+  the API works everywhere and pays off where it matters.
+- **Interceptors see a streamed response without its body** — status and
+  headers, and nothing under them, because nothing was ever assembled.
+
+With retries enabled, a retryable status (429/5xx) is never offered to the
+sink, on any attempt: streaming a 503's error document and then retrying
+would leave the sink holding it followed by the real body, with no way to
+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.
 
 ## Retries
 

docs/research/client-third-party-api-gaps.md

@@ -191,14 +191,24 @@ JSON/blob downloads.
 **6a** for the initial third-party JSON client. Revisit **6b/6d** when a
 concrete multi-MB concurrency budget appears. Do not block 1–4 on **6c**.
 
-**Status (2026-09-09):** 6a, plus the bound it was missing:
-`ClientConfig::max_response_bytes` (default 64 MiB) caps what one buffered
-response can make the process hold, on both built-in transports. Before it
-the Beast client accepted responses of any size. 6b (a transport body sink,
-streaming only for non-retryable statuses so the retry loop stays
-transparent) is the next slice when a consumer has a blob download to make;
-6c waits for a model with a `@streaming` blob, and its upload half needs
-chunked transfer-encoding, which the http1 codec refuses by design.
+**Status (2026-09-11):** 6a and 6b are in; 6c is
+[#213](https://github.com/muchq/opal-cpp/issues/213).
+
+- **6a + the bound it was missing:** `ClientConfig::max_response_bytes`
+  (default 64 MiB) caps what one buffered response can make the process hold,
+  on both built-in transports. Before it the Beast client accepted responses
+  of any size.
+- **6b:** `HttpClient::SendStreaming` takes an `opal::http::BodySink` and
+  hands an accepted body over in pieces; `BeastHttpClient` never holds one,
+  and every other transport inherits a buffering default that honors the same
+  contract. A retryable status is never offered to the sink while retries are
+  enabled, so the retry loop stays transparent. `max_response_bytes` bounds
+  the buffered path only — the cap is about what this process holds.
+- **6c** (a `@streaming` blob behind a generated method) is the remaining
+  work, driven by
+  [MoonBase#1527](https://github.com/muchq/MoonBase/issues/1527). Its upload
+  half needs chunked transfer-encoding, which the http1 codec refuses by
+  design, and is not part of it.
 
 ---
 

examples/bazel-consumer/BUILD.bazel

@@ -125,6 +125,24 @@ 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.
+cc_test(
+    name = "response_sink_acceptance_test",
+    size = "small",
+    srcs = ["response_sink_acceptance_test.cc"],
+    copts = OPAL_COPTS,
+    deps = [
+        ":redirector_client",
+        ":redirector_server",
+        "@googletest//:gtest_main",
+        "@opal_cpp//runtime:core",
+        "@opal_cpp//runtime:http",
+        "@opal_cpp//runtime:server",
+    ],
+)
+
 # Acceptance for ClientConfig::max_response_bytes (issue #189): the
 # generated client, built from an endpoint alone, refuses a blob payload
 # over the budget without retrying and delivers one at the budget whole.

examples/bazel-consumer/response_sink_acceptance_test.cc

@@ -0,0 +1,166 @@
+// 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.
+//
+// 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 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,
+// 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
+// socket-transport tests.
+
+#include <gtest/gtest.h>
+
+#include <cstddef>
+#include <memory>
+#include <string>
+#include <string_view>
+
+#include "acme/redirect/client.h"
+#include "acme/redirect/server.h"
+#include "opal/core/error.h"
+#include "opal/http/message.h"
+#include "opal/http/socket_transport.h"
+#include "opal/http/transport.h"
+
+namespace {
+
+using acme::redirect::FetchInput;
+using acme::redirect::FetchOutput;
+using acme::redirect::NoSuchSlug;
+using acme::redirect::ProbeInput;
+using acme::redirect::ProbeOutput;
+using acme::redirect::RedirectorHandler;
+using acme::redirect::RedirectorServer;
+using acme::redirect::ResolveDynamicInput;
+using acme::redirect::ResolveDynamicOutput;
+using acme::redirect::ResolveInput;
+using acme::redirect::ResolveOutput;
+
+// A payload big enough that nobody would want it in memory twice — the case
+// the sink exists for. The content is a repeating pattern rather than one
+// character so a test can tell a truncated delivery from a short one.
+std::string LargePayload() {
+  std::string payload;
+  payload.reserve(512 * 1024);
+  while (payload.size() < 512 * 1024) {
+    payload += "the quick brown fox jumps over the lazy dog\n";
+  }
+  return payload;
+}
+
+class DownloadHandler final : public RedirectorHandler {
+ public:
+  explicit DownloadHandler(std::string payload) : payload_(std::move(payload)) {}
+
+  opal::Outcome<FetchOutput> Fetch(const FetchInput& input,
+                                   const opal::server::RequestContext&) override {
+    if (input.slug != "big") return NotFound(input.slug);
+    return FetchOutput{
+        .status = 200, .etag = "\"big\"", .content = opal::Blob::FromString(payload_)};
+  }
+  opal::Outcome<ProbeOutput> Probe(const ProbeInput&,
+                                   const opal::server::RequestContext&) override {
+    return ProbeOutput{.etag = "\"big\"", .content = opal::Blob::FromString(payload_)};
+  }
+  opal::Outcome<ResolveOutput> Resolve(const ResolveInput& input,
+                                       const opal::server::RequestContext&) override {
+    return NotFound(input.slug);
+  }
+  opal::Outcome<ResolveDynamicOutput> ResolveDynamic(const ResolveDynamicInput& input,
+                                                     const opal::server::RequestContext&) override {
+    return NotFound(input.slug);
+  }
+
+ private:
+  static opal::Error NotFound(const std::string& slug) {
+    opal::Error error = opal::Error::Modeled("NoSuchSlug", "no slug: " + slug);
+    error.set_detail(NoSuchSlug{.message = "no slug: " + slug});
+    return error;
+  }
+
+  std::string payload_;
+};
+
+class ResponseSinkAcceptanceTest : public ::testing::Test {
+ protected:
+  void SetUp() override { ASSERT_TRUE(transport_.Start(server_.Handler()).ok()); }
+  void TearDown() override { transport_.Stop(); }
+
+  opal::http::HttpRequest Get(const std::string& slug) const {
+    opal::http::HttpRequest request;
+    request.method = "GET";
+    request.target = "/c/" + slug;
+    return request;
+  }
+
+  std::string payload_ = LargePayload();
+  RedirectorServer server_{std::make_shared<DownloadHandler>(payload_)};
+  opal::http::SocketHttpServer transport_;
+};
+
+TEST_F(ResponseSinkAcceptanceTest, ABlobPayloadArrivesThroughTheSinkAndNotInTheResponse) {
+  opal::http::SocketHttpClient client("127.0.0.1", transport_.port());
+
+  // What a real consumer does with the pieces: hand them straight to
+  // something that consumes bytes — a file, a hash, a parser. Nothing here
+  // keeps the payload, only its length and digest.
+  std::size_t received = 0;
+  std::size_t digest = 0;
+  const opal::http::BodySink to_consumer{
+      .accept = [](int status, const opal::http::Headers&) { return status == 200; },
+      .write =
+          [&](std::string_view piece) {
+            received += piece.size();
+            for (const char byte : piece) digest = digest * 31 + static_cast<unsigned char>(byte);
+            return true;
+          },
+  };
+
+  const auto response = client.SendStreaming(Get("big"), to_consumer);
+  ASSERT_TRUE(response.ok()) << response.error().message();
+  EXPECT_EQ(response->status, 200);
+  EXPECT_EQ(response->headers.Get("etag").value_or(""), "\"big\"");
+  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<unsigned char>(byte);
+  EXPECT_EQ(digest, expected) << "the bytes arrived, but not these bytes";
+}
+
+TEST_F(ResponseSinkAcceptanceTest, AModeledErrorIsLeftWhereTheClientLooksForIt) {
+  // The reason accept() is asked per response: a sink that takes payloads
+  // must not swallow the error document that the generated client's own
+  // deserializer needs to turn a 404 into a NoSuchSlug.
+  opal::http::SocketHttpClient client("127.0.0.1", transport_.port());
+
+  bool wrote = false;
+  const opal::http::BodySink payloads_only{
+      .accept = [](int status, const opal::http::Headers&) { return status == 200; },
+      .write =
+          [&](std::string_view) {
+            wrote = true;
+            return true;
+          },
+  };
+
+  const auto response = client.SendStreaming(Get("missing"), payloads_only);
+  ASSERT_TRUE(response.ok()) << response.error().message();
+  EXPECT_EQ(response->status, 404);
+  EXPECT_FALSE(wrote);
+  // The modeled member is still in the document, which is what the generated
+  // client's deserializer reads to build the typed NoSuchSlug detail. Had the
+  // sink taken this response, it would have read an empty body instead.
+  EXPECT_NE(response->body.find("no slug: missing"), std::string::npos) << response->body;
+}
+
+}  // namespace

runtime/BUILD.bazel

@@ -342,6 +342,17 @@ cc_test(
     ],
 )
 
+cc_test(
+    name = "body_sink_test",
+    size = "small",
+    srcs = ["tests/http/body_sink_test.cc"],
+    copts = COPTS,
+    deps = [
+        ":http",
+        "@googletest//:gtest_main",
+    ],
+)
+
 cc_test(
     name = "http_test",
     size = "small",

runtime/include/opal/client/interceptor.h

@@ -25,6 +25,12 @@ class Interceptor {
   // Runs after each attempt with the request as sent and the transport
   // outcome (a response of any status, or a transport error). Observe only:
   // logging, metrics, tracing.
+  //
+  // On a call whose body went to a BodySink (opal/http/transport.h), the
+  // response carries its status and headers and an empty body — the bytes
+  // went to the sink and were never assembled anywhere for a hook to read.
+  // An interceptor that logs response bodies sees nothing there rather than
+  // something truncated.
   virtual void ReadAfterTransmit(const http::HttpRequest& request,
                                  const Outcome<http::HttpResponse>& outcome, int attempt) {
     (void)request;

runtime/include/opal/client/retry.h

@@ -50,6 +50,27 @@ inline Outcome<http::HttpResponse> SendWithRetries(http::HttpClient& transport,
   return SendWithRetries(transport, request, policy, {});
 }
 
+// The same, with the response body streamed to `sink` rather than buffered
+// (issue #213). Retries and streaming disagree about one thing, and this
+// settles it: while retries are enabled, a retryable status (429/5xx) is
+// buffered into HttpResponse::body and never offered to the sink. Streaming a
+// 503's error document and then retrying would leave the sink holding that
+// body followed by the real one, with nothing able to take the first back.
+//
+// The rule is the same on the last attempt as on the first, deliberately: a
+// sink takes payloads, never a transient failure's error document, and which
+// attempt produced a 503 is not something a caller should have to reason
+// about. Nothing is lost either way — such a body is small, and it arrives
+// where every other error document already does. With retries disabled
+// (max_attempts = 1) no response can be discarded, so the caller's accept()
+// stands unaltered for every status.
+//
+// A failure after the sink has taken bytes is not retryable, so the loop ends
+// there whatever the policy says.
+Outcome<http::HttpResponse> SendWithRetries(
+    http::HttpClient& transport, const http::HttpRequest& request, const RetryPolicy& policy,
+    const std::vector<std::shared_ptr<Interceptor>>& interceptors, const http::BodySink& sink);
+
 }  // namespace opal
 
 #endif  // OPAL_CLIENT_RETRY_H_

runtime/include/opal/http/beast_transport.h

@@ -276,12 +276,19 @@ class BeastHttpClient : public HttpClient {
 
   Outcome<HttpResponse> Send(const HttpRequest& request) override;
 
+  // Streams an accepted response body to the sink and never holds it whole
+  // (issue #213), which is what max_response_bytes cannot do for a download
+  // whose size the model does not bound. A declined response is buffered and
+  // capped exactly as Send() buffers and caps it.
+  Outcome<HttpResponse> SendStreaming(const HttpRequest& request, const BodySink& sink) override;
+
  private:
   struct State;  // Hides boost headers from this public header.
 
-  // Send()'s body; Send() wraps it so no exception (a bad_alloc from the sync
-  // drive) crosses the Outcome boundary (ADR-0003).
-  Outcome<HttpResponse> SendContained(const HttpRequest& request);
+  // Both public sends' body (`sink` null for Send()); they wrap it so no
+  // exception (a bad_alloc from the sync drive, or a throwing sink callback)
+  // crosses the Outcome boundary (ADR-0003).
+  Outcome<HttpResponse> SendContained(const HttpRequest& request, const BodySink* sink);
 
   std::shared_ptr<State> state_;
 };