Evan Zou

Wire stream and sink lifecycle logging Summary: Wire `ThriftStreamLog` and `ThriftSinkLog` into the actual stream/sink lifecycle. This diff connects the unified logging layer to the runtime: - Store stream/sink logs as `shared_ptr` on the bridges and callbacks where they are used, rather than on `ContextStack`: - `ServerGeneratorStreamBridge` holds `shared_ptr<ThriftStreamLog>` - `RocketStreamClientCallback` holds `shared_ptr<ThriftStreamLog>` - `ServerSinkBridge` holds `shared_ptr<ThriftSinkLog>` (via `SinkConsumerImpl`) - Pass `ThriftStreamLog` through `ServerStreamFactory` via an additional parameter on `ServerStreamFactoryFn`. - Pass `ThriftSinkLog` through `ServerSinkFactory::StartFunction` into `SinkConsumerImpl`. - In `RocketThriftRequests`, create logs from the connection log and set them on the factory/callback when stream/sink responses are sent. - In `ServerGeneratorStreamBridge`, call `streamLog->log()` at each stream event (subscribe, next, credit, pause, resume, complete, error, cancel). - In `ServerSinkBridge`, call `sinkLog->log()` at each sink event (subscribe, next, consumed, credit, cancel, complete). The default backend is still noop, so no fb303 stats are emitted and no overlap with `StreamMetricEventHandler`. Reviewed By: robertroeser Differential Revision: D96523887 fbshipit-source-id: a70fff390e6525648f135216f7e0b27535d00727

Move InteractionTest.cpp to e2e test directory Summary: Move InteractionTest.cpp (29 tests) from `thrift/lib/cpp2/test/` to `thrift/lib/cpp2/test/e2e/` as InteractionServiceE2ETest.cpp since it tests the full client-server interaction lifecycle. Key changes: - New Thrift IDL `TestInteractionService.thrift` with e2e namespace conventions (`apache.thrift.detail.test`) - New `thrift_cxx_test` target `interaction_service_e2e_test` - All 29 existing tests migrated with updated namespace and removed `#if FOLLY_HAS_COROUTINES` guards - 3 new tests added for coverage gaps: - DeclaredException: verifies declared AdditionException propagation - MultipleConcurrentInteractions: tests state isolation across two Addition interactions on the same client - DifferentInteractionTypesConcurrently: tests Addition + AdditionFast operating independently on the same client - Original Interaction.thrift and InteractionService BUCK target preserved (depended on by other tests) Reviewed By: sazonovkirill Differential Revision: D97845473 fbshipit-source-id: 4179760f63fc5d0745d3e4c4778c324b83596dd6

Migrate BiDi counter reporting to IThriftServerCounters Summary: Migrates bidirectional stream (BiDi) counter reporting from the legacy `ContextStack` → `BiDiEventHandler` → `BiDiMetricEventHandler` path to the new unified `IThriftServerCounters` interface dispatched via `ThriftBiDiLog`. This follows the same architecture as D96523528 which migrated stream/sink counters. A new `ThriftBiDiLog` class (analogous to `ThriftStreamLog` / `ThriftSinkLog`) accumulates per-session state and dispatches to the injected `IThriftServerCounters` for real-time metrics. Key changes: - New `ThriftBiDiLog` class with lifecycle events for subscribe, sink-next, stream-next, credit, and finally - New bidi counter methods on `IThriftServerCounters` with noop defaults - `ThriftConnectionLog::createBiDiLog()` factory returning shared_ptr - `Fb303ThriftServerCounters` implements bidi counters preserving exact same fb303 counter string names for backward compatibility - `ServerBiDiStreamFactory` gains `setMethodName`/`setBiDiLog` and passes the log into bridges and stapler - `ServerBiDiStreamBridge`, `ServerBiDiSinkBridge`, `ServerCallbackStapler` dispatch to `ThriftBiDiLog` alongside existing `ContextStack` calls - `RocketThriftRequests` creates bidi log from connection log (factory path) - `StreamMetricHandler` no longer creates `BiDiState` or `BiDiMetricEventHandler` — bidi metrics handled by the new path Single atomic diff to avoid double-counting: the new path is installed and the old path is removed together. Reviewed By: tlj77 Differential Revision: D98340551 fbshipit-source-id: 98c104baeceae3ee310054156d0d464899c25e02

Added ServiceInterceptor::onRequestDropped Summary: Add `onRequestDropped` callback to the ServiceInterceptor framework, allowing interceptors to observe requests that are dropped before reaching the handler. New API: - `DropReason` enum: UNKNOWN, QUEUE_TIMEOUT, QUEUE_FULL, OVERLOAD, UNKNOWN_METHOD - `DroppedRequestInfo` struct with request context, method name, and drop reason - `onRequestDropped(ConnectionState*, DroppedRequestInfo)` virtual on `ServiceInterceptor` - `processServiceInterceptorsOnDrop()` free function that iterates registered interceptors Wired into the following drop paths: - `ThriftRocketServerHandler`: OVERLOAD (Rocket transport overload rejection) - `Cpp2Worker::handleServerRequestRejection`: UNKNOWN_METHOD / QUEUE_FULL (resource pool rejection) - `Cpp2Worker::dispatchRequest` MetadataNotFound: UNKNOWN_METHOD - `ParallelConcurrencyController`: QUEUE_TIMEOUT (expired requests in executeRequest and processExpiredRequest) - `TokenBucketConcurrencyController`: QUEUE_TIMEOUT (expired requests in release and processExpiredRequest) - `ResourcePool`: QUEUE_TIMEOUT (expired requests in acceptDeferred and acceptSync) The callback is synchronous and may be called from IO or CPU threads depending on the drop reason. The docblock documents threading and ordering constraints. Reviewed By: tlj77 Differential Revision: D100249601 fbshipit-source-id: ec9e812f0d902d0a7d8f6cdedfd117e4083ad31c

Add ThriftSinkLog, ThriftConnectionLog, and unit tests Summary: Add `ThriftSinkLog` (sink counterpart to `ThriftStreamLog`) and `ThriftConnectionLog` (per-connection factory that creates stream/sink logs with injected backends). `ThriftSinkLog` accumulates chunk receive/consume counts, tracks receive intervals, samples consume latency, and approximates client pause duration via credit exhaustion detection. On terminal state, dispatches to `IThriftRequestLogging` with a `SinkSummary`. `ThriftConnectionLog` owns backend references and provides `createStreamLog()` / `createSinkLog()` factory methods that produce correctly-wired log instances. Unit tests cover both classes. Reviewed By: sazonovkirill Differential Revision: D96523886 fbshipit-source-id: d25fc0624b0b214b9eddb8ebf48f7145bb154d34

Add ThriftStreamLog with unit tests Summary: Add `ThriftStreamLog`, a concrete class that accumulates state for a single server-side stream. Each log() overload updates local counters and dispatches to the injected `IThriftServerCounters` for real-time metrics. On terminal state, dispatches to `IThriftRequestLogging` for structured logging. Key behaviors: - Credit accumulation and consumption tracking - Pause/resume state machine with duration measurement - Chunk generation interval (with pause time subtracted) - Send delay sampling (cross-thread coordination via atomics) - Idempotent finish (destructor cleanup if not explicitly completed) Includes comprehensive unit tests covering state accumulation, counter dispatch, terminal events, null backends, destructor cleanup, and idempotent finish. Reviewed By: sazonovkirill Differential Revision: D96523889 fbshipit-source-id: 121abade7dd7e72631ac54fe9f2f5d6572e726d0

Add stream credit starvation timeout to RocketBiDiClientCallback Summary: RocketStreamClientCallback has a stream credit starvation timeout that tears down the stream if the client stops sending REQUEST_N credits after the server exhausts its tokens (default 60s). RocketBiDiClientCallback lacked this protection. This adds the same timeout mechanism to bidi streams: - Track `streamTokens_` consumed by `onStreamNext()` and replenished by incoming `REQUEST_N` - Schedule timeout via `connection_.scheduleStreamTimeout()` when tokens reach zero - Cancel timeout when new credits arrive, stream completes, or connection closes - On timeout expiry, cancel both stream and sink halves and send `CREDIT_TIMEOUT` error Reviewed By: sazonovkirill Differential Revision: D97152008 fbshipit-source-id: 699ecb08ed313636e75435389bdf1300fbee504d

Introduce IConnectionStreamHandler interface and implement on RocketStreamClientCallback Summary: Introduce `IConnectionStreamHandler` — a connection-facing interface with typed `handleFrame` overloads for each frame type (RequestN, Cancel, Payload, Error, Ext). This is the first step toward replacing `variant_match` dispatch in `RocketServerConnection` with virtual calls. This diff: - Creates the `IConnectionStreamHandler` pure interface (header-only) - Implements the interface on `RocketStreamClientCallback`: - `handleFrame(RequestNFrame&&)` — pre-ready check, delegates to `handle()` - `handleFrame(CancelFrame&&)` — delegates to `handle()` (allowed pre-ready) - `handleFrame(ExtFrame&&)` — pre-ready check, delegates to `handle()` - `handleFrame(PayloadFrame&&)` — closes connection (unsupported for streams) - `handleFrame(ErrorFrame&&)` — closes connection (unsupported for streams) - Updates `handleConnectionClose()` to also call `sendErrorAfterDrain()` (logic previously inline in `RocketServerConnection::closeIfNeeded`) - Moves `getStreamConnectionClosingError()` helper into the callback - Creates `MockIRocketServerConnection` for unit testing callbacks - Adds 9 unit tests for the stream callback's `IConnectionStreamHandler` methods No callsite changes — the old `variant_match` dispatch is untouched. Reviewed By: sazonovkirill Differential Revision: D94946702 fbshipit-source-id: 7611d078ded463fb06fbbaf7b8391c76088826d7

Implement IConnectionStreamHandler in RocketBiDiClientCallback Summary: Implement `IConnectionStreamHandler` on `RocketBiDiClientCallback`, the third callback type (after stream and sink). This absorbs the frame dispatch logic from `RocketServerConnection::handleBiDiFrame` into typed `handleFrame` overloads on the callback itself: - `PayloadFrame`: fragment reassembly via private `bufferOrGetFullPayload()`, then dispatches to `onSinkNext`/`onSinkComplete`/`onSinkError` based on flags and exception metadata - `ErrorFrame`: pre-ready CANCELED → `cancelEarly()`, post-ready → `onSinkError` with appropriate exception wrapper - `RequestNFrame`: delegates to `onStreamRequestN` if stream is open - `CancelFrame`: delegates to `onStreamCancel` - `ExtFrame`: closes connection if `!hasIgnore()` - `handleConnectionClose`: notifies server callback of both sink error and stream cancel Also adds `bufferOrGetFullPayload()` private helper and `bufferedFragment_` member for per-callback fragment reassembly (same pattern as sink callback), and makes `serverCallbackReady()` const. Reviewed By: sazonovkirill Differential Revision: D94946937 fbshipit-source-id: 5f0934a019ab4e47c1ffd9354c85b7deed408e31

Replace variant dispatch with IConnectionStreamHandler virtual dispatch Summary: Replace all `variant_match` dispatch in `RocketServerConnection` and `RefactoredRocketServerConnection` with virtual dispatch through the `IConnectionStreamHandler` interface. Key changes: - Change `ClientCallbackUniquePtr` from `std::variant<unique_ptr<Stream>, unique_ptr<Sink>, unique_ptr<BiDi>>` to `unique_ptr<IConnectionStreamHandler>` in both `IRocketServerConnection.h` and `RefactoredRocketServerConnection.h` - Replace frame dispatch `variant_match` in `RocketServerConnection::handleFrame` with a switch/case that constructs typed frames and calls `handler->handleFrame()` - Replace `closeIfNeeded` stream cleanup with `handler->handleConnectionClose()` - Replace `pauseStreams`/`resumeStreams` with `handler->handlePausedByConnection()` / `handleResumedByConnection()` - Replace METADATA_PUSH dispatch with `handler->handleStreamHeadersPush()` - Remove `handleSinkFrame`, `handleBiDiFrame`, `bufferOrGetFullPayload` methods from both connections (~540 lines of dead code) - Remove shared `bufferedFragments_` map (each callback now owns its buffering) - Update detail/ helpers: `ExistingStreamFrameHandler`, `StreamCallbackManager`, `MetadataPushHandler`, `ConnectionAdapter` Reviewed By: sazonovkirill Differential Revision: D94946960 fbshipit-source-id: 05b76936c50463d1072ddcbddebdac62a88a8a06

Add sink chunk timeout to RocketBiDiClientCallback Summary: RocketSinkClientCallback has a per-method chunk timeout that tears down the sink if the client stops sending payloads after the server grants credits. RocketBiDiClientCallback lacked this protection. This adds the same timeout mechanism to bidi streams: - Add `chunkTimeout` field to `StreamTransformation` (user-facing API) - Thread `chunkTimeout` through `encode_server_bidi_stream` → `ServerBiDiStreamFactory` - Add `SinkTimeoutCallback` inner class that tracks sink credits and schedules/cancels the timeout via `connection_.scheduleSinkTimeout()` - Wire up `setChunkTimeout()` in `RocketThriftRequests.cpp` - On timeout expiry, notify server of sink error, cancel stream if open, send `CHUNK_TIMEOUT` error to client, and free the stream Reviewed By: sazonovkirill Differential Revision: D97152042 fbshipit-source-id: 48650fc23f987c6ee7103320f882abe8f0c922e0

Implement IConnectionStreamHandler on RocketSinkClientCallback Summary: Implement `IConnectionStreamHandler` on `RocketSinkClientCallback`, absorbing sink frame dispatch logic from `RocketServerConnection::handleSinkFrame`. This diff: - `RocketSinkClientCallback` inherits from `IConnectionStreamHandler` - Adds `streamId()` override - Adds `bufferOrGetFullPayload()` private helper and `bufferedFragment_` member to own payload fragment reassembly (previously shared state on connection) - New `handleFrame` overloads: - `PayloadFrame` — fragment reassembly, then onSinkNext/onSinkComplete/onSinkError - `ErrorFrame` — pre-ready CANCELED → earlyCancelled(), post-ready → onSinkError - `RequestNFrame`, `CancelFrame`, `ExtFrame` — close connection (unsupported) - New `handleConnectionClose()` — calls onSinkError with INTERRUPTION - Adds 8 unit tests for the sink callback's IConnectionStreamHandler methods No callsite changes — the old variant_match dispatch is untouched. Reviewed By: sazonovkirill Differential Revision: D94946716 fbshipit-source-id: c0d097451892214a555b26abfc871e5c9e5ef0cd

Add onRequestReceived hook to ServiceInterceptor Summary: Add a new `onRequestReceived` hook to `ServiceInterceptor` that fires synchronously for every request that reaches the central dispatch point (`Cpp2Worker::dispatchRequest`), before any admission or rejection decisions. This provides a single observation point for all incoming requests. A request that triggers `onRequestReceived` will subsequently receive exactly one of `onRequest` (processed) or `onRequestDropped` (rejected). This is useful for metrics, logging, and tracing that need to observe all requests regardless of outcome. Changes: - Added `ReceivedRequestInfo` struct and `internal_onRequestReceived` virtual method to `ServiceInterceptorBase` - Added typed `onRequestReceived` virtual method to `ServiceInterceptor<RequestState, ConnectionState>` - Created `ServiceInterceptorOnReceived.h/.cpp` with `processServiceInterceptorsOnReceived` dispatch function - Wired the hook at the top of `Cpp2Worker::dispatchRequest`, which is the convergence point for all transports (Rocket, Header, HTTP) Reviewed By: robertroeser Differential Revision: D101725063 fbshipit-source-id: 9a2d0398d1c5d368f93db539ad9b35d579602feb

Add ODS counter metrics for BiDi stream pause/resume events Summary: Add pause/resume counter dispatch to ThriftBiDiLog, mirroring the existing ThriftStreamLog pattern. This enables ODS visibility into bidi stream backpressure behavior. Changes: - Add BiDiStreamResumeEvent to ThriftEvent.h - Add onBiDiStreamPause and onBiDiStreamResume to IThriftServerCounters - Implement pause/resume state tracking in ThriftBiDiLog (pause reason, duration, event count) with counter dispatch on each transition - Adjust bidi stream generation interval to subtract pause duration, matching ThriftStreamLog behavior - Log BiDiStreamResumeEvent from ServerBiDiStreamBridge's Resume handler - Clean up paused state on bidi finish (handleResume in finish()) Reviewed By: sazonovkirill Differential Revision: D100030647 fbshipit-source-id: ab246f384bee146060791c8d43f30ee21c680b9f

Add genWaitBiDi hook to TClientAsyncHandler for bidirectional streaming Summary: Add genWaitBiDi to TClientAsyncHandler so that handler-based clients (like FaaS stateful) can intercept bidi RPCs. The handler returns a tuple of (stream_generator, sink_function) operating on raw serialized bytes, matching the pattern of genWaitStream and genWaitSink. Update ThriftClientBase::genAwaitBiDiStreamResponse to fall back to the async handler when no channel is present, mirroring the existing pattern in genAwaitStreamResponse. Reviewed By: rmakheja Differential Revision: D104442817 fbshipit-source-id: 965ab5256b8cd1220cb153907462a31f28b3bc7d

Add coroutine cancellation support for BiDi streaming bridges Summary: When the client cancels a BiDi stream while the server transform is blocked on `co_await input.next()` (waiting for sink data), the Cancel message sits unprocessed in the stream bridge queue and the coroutine hangs indefinitely. This adds the same cancellation patterns already used by `ServerGeneratorStreamBridge` (streams) and `ServerSinkBridge` (sinks): **ServerBiDiStreamBridge (stream direction):** - Add `cancelSource_` member fired in `onStreamCancel()` - Wrap `input.next()` with `co_withCancellation` in `getTask()` - Handle `OperationCancelled` by returning without pushing an error (client side is already closed by `onStreamCancel`) **ServerBiDiSinkBridge (sink direction):** - Add `CancellationCallback` in `getInput()` that calls `serverClose()` when cancellation is requested, unblocking the `CoroConsumer` wait - Add `co_safe_point` after the wait to throw `OperationCancelled` - Guard `Cleanup` destructor with `isServerClosed()` to prevent double-close of the `AtomicQueue` (which triggers `assume_unreachable`) Reviewed By: sazonovkirill Differential Revision: D94977350 fbshipit-source-id: 8dc2b3a8c9328f04454db76f0869065b5913a496

Fix use-after-free of serverCallback_ in RocketBiDiClientCallback Summary: Fix two lifetime issues with the raw `serverCallback_` pointer in `RocketBiDiClientCallback`: 1. **`handleConnectionClose` double-call**: The method calls `serverCallback_->onSinkError()` then `serverCallback_->onStreamCancel()` back-to-back. If the bridge's `onSinkError` destroys the server callback, the second call crashes on a dangling pointer. Fix: use `std::exchange(serverCallback_, nullptr)` before making the calls, routing them through a saved local instead. 2. **Terminal state not freeing stream**: When an incoming method (`onSinkError`, `onSinkComplete`, `onStreamCancel`) closes the last open direction, the state becomes terminal. Previously, if `serverCallback_->on...()` returned true (no contract violation), the code hit `DCHECK(state_.isAlive())` (fatal in debug) and returned `true` (wrong) in release — leaving the stream un-freed and `serverCallback_` dangling for subsequent frames. Fix: when `wasTerminal` is true after the server callback call succeeds, null out `serverCallback_` and call `freeStreamAndReturn`. The SIGSEGV at address 0x28 in production was caused by issue 2: the server-side bridge completed and destroyed itself while the BiDi stream remained in the connection's `streams_` map. A subsequent ERROR frame from the client passed the `serverCallbackReady()` check (non-null pointer) and `isSinkOpen()` check (state not updated), then crashed calling virtual `onSinkError` through the zeroed vtable of the freed object (0x28 = vtable slot offset for `onSinkError`). Reviewed By: sazonovkirill Differential Revision: D98339564 fbshipit-source-id: aca2eafef7e8d9cfdf25a0e2b3cf41a82846a228

Add Pause/Resume support to ServerBiDiStreamBridge Summary: Extend the bidi stream bridge to handle Pause/Resume messages from the transport layer, mirroring the existing support in ServerGeneratorStreamBridge for unidirectional streams. This is the foundational change for egress backpressure in bidirectional streams. Changes: - Add StreamMessage::Pause and StreamMessage::Resume to the ServerBiDiStreamMessageClientToServer variant type - Override pauseStream() and resumeStream() on ServerBiDiStreamBridge to push Pause/Resume messages through the TwoWayBridge - Update getTask() coroutine to handle Pause/Resume messages: when paused, the coroutine blocks until a Resume is received, even if credits are available Reviewed By: sazonovkirill Differential Revision: D99339318 fbshipit-source-id: 2394b812bed59131ddd0840b9ab0d2cb32a51647

Add bidirectional streaming support to GuardedRequestChannel Summary: `GuardedRequestChannel` wraps a `RequestChannel` with RAII guards for fork safety. It overrides `sendRequestResponse`, `sendRequestNoResponse`, `sendRequestStream`, and `sendRequestSink` but not `sendRequestBiDi`, causing the fork-safe path to fall through to the base class `terminate_with`. This diff adds: - `GuardedBiDiCallback<RequestGuardType>`: implements both `BiDiClientCallback` and `BiDiServerCallback`, holds the RAII guard for the duration of the bidi session, and tracks both stream and sink halves independently via `streamAlive_`/`sinkAlive_` flags. The callback only self-deletes when both halves are closed. The destructor unstaples by reconnecting client and server callbacks directly. - `sendRequestBiDi` override on `GuardedRequestChannel` that wraps the client callback in `GuardedBiDiCallback` before delegating to the inner channel. Reviewed By: sazonovkirill Differential Revision: D101887555 fbshipit-source-id: 19addf0d4502a62980e5cc6e5bdbf3b18a5b8f35

Add server stream benchmark to ServiceLab Summary: Add a new ServiceLab benchmark that measures end-to-end server stream performance over the Rocket transport. This fills a gap in the existing benchmark coverage — the current suite covers request-response (RpcBench, ClientBench), serialization (protocol_bench), and field access patterns, but has no coverage for streaming RPCs. The benchmark measures three dimensions: - **Stream element count** (1, 100, 10,000): measures setup overhead through sustained throughput - **Payload size** (64B, 1KB, 64KB at 100 elements): measures serialization and buffer management cost - **Server-side work** (100, 10,000 iterations at 100 elements): measures behavior under compute-bound server handlers The benchmark creates a real ThriftServer via ScopedServerInterfaceThread, implements handlers that co_yield values, and consumes them on the client side via AsyncGenerator. The ServiceLab entry is configured as an active landblock guard (20% threshold), triggering on both diffs and lands for changes under thrift/lib/cpp2. Reviewed By: sazonovkirill Differential Revision: D97391197 fbshipit-source-id: e9bf402c34f98fecd922f42135b36adaf5e39a1b