feat: Pluggable transport response shaping inversion

[sichanyoo]

Followup to:

• feat: Pluggable transport interface + NetHTTP default transport. #357

Issue #, if available:
5196

---

Description of changes:

The summary below was written with help of AI.

Invert response shaping into the transport (push/sink model)

What changed at a glance

The previous PR (#357) landed the pull Transport/Stream contract: transport.transmit(request) returns a Stream, and the SDK pulls the response off it (stream.response_headers, then stream.each_chunk). This PR replaces that with a push (sink) model and splits the send surface by operation mode.

• transmit(request) → transmit(request, sink) and a new transmit_background(request, sink).
• The SDK no longer pulls the response; the transport pushes it into a caller(i.e., SDK send handler)-supplied sink (sink.headers → sink.data* → one terminal sink.done/sink.error).
• The returned Stream is now a pure control handle (event streams only); a new internal NetHTTP::Exchange holds the driving engine.
• The Net::HTTP Fiber is deleted.

No change to generated client options or any public client API. Suite: 897 examples, 0 failures, 2 pending (the 2 pending are the pre-existing net_http/patches_spec skips on net-http >= 0.7.0); rubocop clean.

Why

You'll remember the two concurrency bugs fixed during the pluggable-transport PR — the abort/session-ownership race and the fiber-local Thread#[] content-type flag. Both traced to the same root cause: Net::HTTP is push-native (http.request(req) { |resp| resp.read_body { |chunk| ... } }), but the pull Stream contract forced us to fake pull by wrapping the exchange in a Fiber and Fiber.yield-ing one chunk per each_chunk. That Fiber was the bug source. The big picture problem of this is that this forces every 3rd party implementer that wants to use a push-based http library to force-fit it into SDK's pull-based interface, which is prone to bugs.

If the transport instead pushes chunks into a sink, Net::HTTP's own block callback maps straight onto sink.data(chunk) — no Fiber, and the entire class of fiber-related concurrency bugs becomes structurally impossible. That's the core motivation; the two-method split and the Exchange/Stream separation fall out of doing it cleanly.

If SDK provides sink to push into, a custom implementation that uses push-based http library is trivial (just map it to sink methods), and a custom implementation that uses pull-based http library is also pretty convenient; it would just pull from transport and push into sink using sink methods. Unlike the pull based interface (where SDK "pulls" from transport) that causes push-based http library to have to go through painful push=>pull adaptor, push based interface provides painless interface for both push and pull based http libraries.

The new contract (lib/smithy-client/transport.rb, stream.rb)

Transport — two send methods, one per operation mode (mode is event-stream vs. non-event-stream and it comes from the service model, not the wire protocol):

• #transmit(request, sink) — non-event-stream ops. Drives the response into the sink to its terminal synchronously, returns nothing; the transport owns teardown (no handle to abort nor write to). Networking failures → sink.error(NetworkingError); invalid verb → ArgumentError synchronously before any I/O.
• #transmit_background(request, sink) — event-stream ops. Starts the exchange concurrently (transport-owned mechanism) and returns a Stream handle immediately.
• #event_queue — the concurrency-appropriate push→pull bridge queue for the (future) event-stream layer. Concretely, this bridges the [transport's push to sink] and the [SDK's event receiver pulling and enumerating events] to the SDK user. We leave this up to transport implementation because the method for allowing this concurrent access depends on the concurrency mechanism used by the transport (e.g., for Net::HTTP it's SizedQueue, for async client based on async gem, it's Async::LimitedQueue, etc.).
• REQUIRED_METHODS = %i[transmit transmit_background] — both required. A single-mode transport implements both and raises NotSupportedError from the one it doesn't serve (mirroring Stream#write/#close_write on H1). This is deliberately symmetric — neither mode is a privileged "base."

Stream — now the outbound + control handle returned only by transmit_background. Inbound no longer flows through it. Capability tiers:

• REQUIRED_FOR_OUTPUT_EVENT_STREAM_OPS = [:abort]
• REQUIRED_FOR_BIDI_EVENT_STREAM_OPS = [:abort, :write, :close_write]

NetHTTP: Exchange (new) vs Stream (thin)

• net_http/exchange.rb (new) — the request/response driving engine: drive/drive_background/abort, plus run/perform_exchange/push_response, the session-ownership handoff, and truncated-body detection. This is where the bulk of the old request-response logic moved to. Semantics preserved exactly: abort/session handoff and truncated-body detection are the same mechanisms you reviewed before — re-run the abort + connection_pool + truncation specs as the acceptance gate.
• net_http/stream.rb — shrinks from the ~340-line Fiber stream to a ~50-line handle wrapping an Exchange: abort delegates to Exchange; write/close_write raise NotSupportedError (H1 is not bidirectional). No inbound data handling (removed each_chunk, etc.).
• net_http/transport.rb — transmit drives an Exchange inline (returns nil); transmit_background backgrounds one (Thread.new { drive }) and wraps it in a Stream and returns it as a handle to caller.

SDK wiring

• send_handler.rb — builds a ResponseSink and selects transport method to call by mode: transmit_background + store the handle when context[:event_stream], else transmit (no stored handle, no ensure-abort — the transport owns teardown). SendHandler is no longer handles feeding data into response.
• response_sink.rb (new) — thin adapter forwarding headers/data/done/error onto Http::Response#signal_*. The signal_*/on_* event bus is unchanged (the event bus / listener API for response is not removed in this PR, it will be added in a followup PR after this one).
• Context flag rename: context[:duplex_stream] → context[:event_stream]

Concurrency review (worth your attention)

The abort path got hardened, since transmit_background means abort runs on a different thread than the driver:

• No sink delivery after abort: every sink.headers/sink.data goes through Exchange#deliver, which checks @aborted and calls the sink atomically under @mutex — so nothing is delivered once an abort is recorded (previously relied on the socket close eventually breaking the read loop).
• mark_done now nils @session, making "done ⇒ nothing left to finish" a hard invariant rather than an ordering coincidence.
• ConnectionPool#finish_session(session, endpoint = nil) — targeted delete when the endpoint is known (O(sessions-at-endpoint) instead of scanning the whole global pool); removal + finish remain atomic under @pool_mutex (an off-lock variant was tried and reverted because it moved the abort-vs-checkin guarantee onto the caller).

One honest limitation: the guarantee is "no delivery after an abort is recorded." A concurrent abort still racing to acquire @mutex can be beaten by one in-flight chunk already inside deliver. That's inherent to not blocking the reader thread, and is the intended semantics.

Scope / what's NOT here

• No event-stream layer. context[:event_stream] is not set by any production code yet (rpc_v2_cbor only uses event_stream? for the Content-Type/Accept headers), so transmit_background/Stream/the capability tiers are forward-compat scaffolding exercised only by tests today. There's a TODO for the event-stream plugin to add the call-time fail-fast for an event op against an event-incapable transport (respond_to? can't detect a NotSupportedError-raising stub).
• TruncatedBodyError stays internal — nested in Exchange, < IOError (same as V3's Seahorse handler), wrapped into NetworkingError before it reaches callers (so it retries as a transient). A test pins that customers can still distinguish it via NetworkingError#original_error.

Tests

• New net_http/exchange_spec.rb (driving, abort, truncation, HEAD, drive_background, and no-delivery-after-abort).
• net_http/stream_spec.rb rewritten to test only the handle (abort delegation, NotSupportedError, contract tiers).
• Shared compliance examples support/transport_contract.rb / stream_contract.rb updated to the push contract; support/recording_sink.rb (new) records the pushed lifecycle. net_http/transport_spec.rb now leans on it_behaves_like 'a transport' (removed same-class re-tests) and adds an event_queue spec.

Suggested review order

1. lib/smithy-client/transport.rb + stream.rb — the contract.
2. net_http/exchange.rb — confirm the abort/session/truncation semantics match what you reviewed on main (this is where they moved), and scrutinize #deliver/#abort/run.
3. net_http/stream.rb + net_http/transport.rb — the thin handle + two-method wiring.
4. send_handler.rb + response_sink.rb — SDK-side wiring; note the event bus is untouched.
5. connection_pool.rb — the finish_session change.

---

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.