fbthrift — Engineering Performance

Add synchronous Rust handler integration Summary: Build the synchronous Rust handler bridge on top of the native channel_pipeline readiness primitives. Add borrowed callback contexts, message adapters, lifecycle and backpressure handling, panic containment, tests, benchmarks, and design documentation while keeping Rust dependencies opt-in. Reviewed By: pranavtbhat Differential Revision: D113432296 fbshipit-source-id: d14a9b4489ea136c34890b8b84e94eabebf95593

Move `thrift/facebook/json5/*` to `thrift/lib/cpp2/protocol` Summary: This diff moved all files under `thrift/facebook/json5 to `thrift/lib/cpp2/protocol`: * `Json5Protocol.h` is moved to `thrift/lib/cpp2/protocol`. Other files are moved to `thrift/lib/cpp2/protocol/detail`. * `thrift/facebook/json5/test` is moved to `thrift/lib/cpp2/protocol/test` * `thrift/facebook/json5/test/example.thrift` is renamed to `json5_test.thrift` Reviewed By: aristidisp Differential Revision: D95818321 fbshipit-source-id: 19d443e93b3a5ea5965b2e62893407874c9634e2

Add shared-parser Thrift formatter binary Summary: Add the `thrift-fmt` binary and formatter library using the shared recursive-descent parser core for grammar validation and token collection from the first formatter diff. Formatter-specific trivia handling and layout remain in the formatter implementation; the semantic AST is not used as a formatting transport. Port the existing formatter snapshot coverage into C++ and add fixture round-trip/idempotence coverage plus the fbcode compatibility target. This implementation exactly preserves the behavior of the existing formatter, including all of its deficiencies, to facilitate deployment. The goal is to fix these issues individually and end up with a better formatter, using the flexibility of controlling the whole stack to make the behavior fully configurable while also eliminating divergence caused by parser duplication as the language evolves. Reviewed By: hchokshi, sadroeck Differential Revision: D108951643 fbshipit-source-id: d00e3ec3d0976ece612e7186c842257aa4ab0a2f

Whitespace control Summary: Add tilde whitespace trimming (`{{~ ~}}`) to the Whisker template language, giving template authors fine-grained control over whitespace in generated code. Whisker templates generate source code across 12 languages, but producing correctly-formatted output has required pervasive workarounds. **282+ template files** use the `{{!` comment hack — empty comments inserted solely to suppress unwanted newlines and whitespace. The result is templates like this, where the actual logic is buried under noise: ```mustache #[derive({{! }}{{#if struct:copy?}}Copy, {{/if struct:copy?}}{{! }}Clone, PartialEq{{! }}{{#if struct:ord?}}, Eq, PartialOrd, Ord, Hash{{/if struct:ord?}}{{! }}{{#if struct:serde?}}, ::serde_derive::Serialize{{/if struct:serde?}}{{! }}) ``` With tilde trimming, the same template becomes readable at a glance: ```mustache #[derive( {{~ #if struct:copy? }}Copy, {{/if struct:copy? ~}} Clone, PartialEq {{~ #if struct:ord? }}, Eq, PartialOrd, Ord, Hash{{/if struct:ord? ~}} {{~ #if struct:serde? }}, ::serde_derive::Serialize{{/if struct:serde? ~}} )] ``` See whisker.md for more details. Reviewed By: hchokshi Differential Revision: D97992973 fbshipit-source-id: 9a9d6a9738fc6fdfa53f8467c1c7205c65defd24

Improve MonoTimeoutTransformer's reactive compliance and multithreaded safety Summary: # MonoTimeoutTransformer: Complete Rewrite for Reactive Streams Compliance ## Summary: This diff is a complete rewrite of MonoTimeoutTransformer to fix multiple race conditions, ensure Reactive Streams specification compliance, and add comprehensive test coverage. ## Why This Rewrite Was Necessary The original implementation had fundamental design flaws: 1. **No explicit state machine**: Used `isDisposed()` checks which are not atomic with respect to signal emission, allowing races. 2. **Incorrect fallback subscription**: Subscribed fallback directly to `actual::onNext, actual::onError, actual::onComplete` method references, bypassing cancellation checks entirely. 3. **Race conditions**: Multiple unhandled races between: - Source emission vs timeout firing (Race 1) - Cancel vs fallback subscription setup (Race 2) - Cancel after timeout but before scheduler execution (Race 3) 4. **No timer cleanup verification**: Timer cancellation on normal completion was not guaranteed. 5. **Untestable**: Timer was hardcoded, making deterministic race testing impossible. ## The Three Critical Race Conditions ### Race 1: Source vs Timeout (Signal Race) ``` Source Thread Timer Thread │ │ │── onNext() ──────────────────────────│── run(Timeout) ── │ [want to emit value] │ [want to emit error] │ │ └──────────── WHO WINS? ───────────────┘ ``` **Old behavior**: Both could win, potentially emitting both value and error. **New behavior**: Atomic CAS on STATE ensures exactly one winner. ### Race 2: Cancel vs Fallback Setup ``` Scheduler Thread Downstream Thread │ │ │─ fallback.subscribe(actual) ────────────────▶│ │ │── cancel() ── │ │ [too late?] ``` **Old behavior**: Fallback subscription ignored cancellation. **New behavior**: FallbackSubscriber uses Operators.set() to atomically check for cancellation before accepting the subscription. ### Race 3: Cancel After Timeout, Before Scheduler (TIMEOUT_MARKER Fix) ``` Timer Thread Downstream Thread Scheduler Thread │ │ │ │─ run(Timeout) ──────────▶│ │ │ [S → ???] │ │ │ [schedule(this)] │ │ │ │ │ │ │── cancel() ──────────────▶│ │ │ [Operators.terminate(S)]│ │ │ [did it work?] │ │ │ │ │ │ │─ run() ── │ │ │ [should skip?] ``` **Old behavior (if it existed)**: Would use `cancelledSubscription()` as marker in run(Timeout). But `cancel()` also uses `cancelledSubscription()`, so `Operators.terminate()` would see S as already cancelled and do nothing. The scheduler's `run()` would then proceed to emit signals to a cancelled subscriber. **New behavior**: Introduced `TIMEOUT_MARKER` - a distinct sentinel that marks "timeout fired, transitioning to fallback." This allows: 1. `run(Timeout)` sets S to TIMEOUT_MARKER (not cancelledSubscription) 2. If `cancel()` is called, `Operators.terminate()` successfully changes S from TIMEOUT_MARKER to cancelledSubscription() 3. `run()` uses CAS to check if S is still TIMEOUT_MARKER - if not, cancellation occurred and we skip signal emission ## Architecture Overview ### Old Architecture (Flawed) ``` ┌─────────────────────────────────────────────────────────────────────────────┐ │ MonoTimeoutTransformer (Old) │ │ │ │ ┌─────────────┐ ┌──────────────────────┐ ┌─────────────────────────┐ │ │ │ Source │───▶│ TimeoutSubscription │───▶│ Downstream │ │ │ │ Mono │ │ (just a holder) │ │ (CoreSubscriber) │ │ │ └─────────────┘ └──────────┬───────────┘ └─────────────────────────┘ │ │ │ │ │ │ creates │ │ ▼ │ │ ┌──────────────────────┐ │ │ │ SourceSubscriber │ ◀── extends BaseSubscriber │ │ │ (no state machine) │ (uses isDisposed()) │ │ └──────────────────────┘ │ │ │ │ Problems: │ │ - No atomic state management │ │ - Fallback subscribed via method references (no cancel check) │ │ - Race conditions between timeout/source/cancel │ └─────────────────────────────────────────────────────────────────────────────┘ ``` ### New Architecture (Correct) ``` ┌─────────────────────────────────────────────────────────────────────────────────┐ │ MonoTimeoutTransformer (New) │ │ │ │ ┌─────────────┐ ┌───────────────────┐ ┌────────────────────────────┐ │ │ │ Source │─────▶│ TimeoutSubscriber │─────▶│ Downstream │ │ │ │ Mono │ │ (CoreSubscriber,│ │ (CoreSubscriber) │ │ │ └─────────────┘ │ Subscription, │ └────────────────────────────┘ │ │ │ TimerTask, │ │ │ │ Runnable) │ │ │ └────────┬──────────┘ │ │ │ │ │ ┌────────────────────┼────────────────────┐ │ │ │ │ │ │ │ ▼ ▼ ▼ │ │ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────────────────────┐ │ │ │ STATE (atomic) │ │ S (atomic) │ │ FallbackSubscriber │ │ │ │ INIT=0 │ │ source sub │ │ (defensive cancel checks) │ │ │ │ VALUE_EMITTED=1│ │ TIMEOUT_MARKER │ │ - checks S before onNext │ │ │ │ TERMINATED=2 │ │ fallback sub │ │ - checks S before onError │ │ │ └─────────────────┘ │ CANCELLED │ │ - checks S before onComplete │ │ │ └─────────────────┘ └─────────────────────────────────┘ │ │ │ │ Key invariants: │ │ - STATE transitions are atomic (CAS) │ │ - Only one thread can win INIT → VALUE_EMITTED or INIT → TERMINATED │ │ - S tracks active subscription and cancellation state │ │ - TIMEOUT_MARKER enables Race 3 detection │ └─────────────────────────────────────────────────────────────────────────────────┘ ``` ## State Machine ``` ┌───────────────────────────────────────────────────┐ │ │ ▼ │ ┌──────────┐ │ │ INIT │◀─────────────────────────────────────────────┘ │ (0) │ (initial state) └────┬─────┘ │ ┌─────────┴─────────┬──────────────────┬──────────────────┐ │ │ │ │ │ onNext() │ timeout fires │ onComplete() │ cancel() │ [CAS succeeds] │ [CAS succeeds] │ [CAS succeeds] │ [getAndSet] ▼ │ │ │ ┌─────────────┐ │ │ │ │VALUE_EMITTED│ │ │ │ │ (1) │ │ │ │ └──────┬──────┘ │ │ │ │ │ │ │ │ onComplete() │ │ │ │ [CAS succeeds] │ │ │ ▼ ▼ ▼ ▼ ┌─────────────────────────────────────────────────────────────┐ │ TERMINATED (2) │ │ (terminal state - no further state transitions allowed) │ └─────────────────────────────────────────────────────────────┘ ``` ## Key Implementation Changes ### 1. Atomic State Management (Race 1 Fix) The original code had no coordination between source signals and timeout. Both could proceed simultaneously, potentially delivering multiple terminal signals. **Before - Source emits value:** ```java Override protected void hookOnNext(T value) { timeout.cancel(); // Cancel timer, but might be too late actual.onNext(value); // Emit value unconditionally } ``` **Before - Timeout fires:** ```java private void doTimeout(Timeout timeout) { if (!isDisposed() && !timeout.isCancelled()) { // Check disposed doTimeout(); // But source might emit between check and here! } } private void doTimeoutException() { TimeoutException e = new TimeoutException("..."); scheduler.schedule(() -> actual.onError(e)); // Emit error } ``` **The Race:** If source emits while timeout fires: 1. `hookOnNext` calls `timeout.cancel()` - but timer callback already started 2. `doTimeout` checks `isDisposed()` → false (source hasn't finished yet) 3. Both `actual.onNext(value)` AND `actual.onError(e)` can execute 4. Downstream receives BOTH value and error → violates Reactive Streams spec **After - Both paths compete via atomic CAS:** ```java Override public void onNext(T t) { // CAS: Only ONE thread can transition INIT → VALUE_EMITTED if (STATE.compareAndSet(this, STATE_INIT, STATE_VALUE_EMITTED)) { cancelTimer(); actual.onNext(t); } else { // Lost the race - drop the value Operators.onNextDropped(t, actual.currentContext()); } } Override // TimerTask.run - called by timer thread public void run(Timeout timeout) { // CAS: Only ONE thread can transition INIT → TERMINATED if (STATE.compareAndSet(this, STATE_INIT, STATE_TERMINATED)) { // Won the race - cancel source and emit timeout Subscription current = S.getAndSet(this, TIMEOUT_MARKER); if (current != null) current.cancel(); scheduler.schedule(this); // Will emit error or subscribe fallback } // If CAS fails, source already won - do nothing } ``` **Why this works:** The CAS operation is atomic. Exactly one of source or timeout can successfully change STATE from INIT. The loser's CAS returns false, and they drop their signal. ### 2. Atomic Subscription Management (Race 2 & 3 Fix) The original code had no atomic tracking of subscription state. The new code uses an atomic `S` field to track the active subscription and detect cancellation. **Before:** ```java private static class TimeoutSubscription<T> implements Subscription { private SourceSubscriber<T> sourceSubscriber; // Not atomic! Override public void cancel() { if (sourceSubscriber != null) { sourceSubscriber.dispose(); // Just sets a boolean } } } ``` **Problem:** `dispose()` sets an internal boolean, but this isn't atomic with respect to subscription switching. If cancel() races with fallback subscription setup: - `dispose()` might set the boolean after fallback checks it - Or before fallback even exists **After:** ```java volatile Subscription s; static final AtomicReferenceFieldUpdater<TimeoutSubscriber, Subscription> S = ...; // TIMEOUT_MARKER: distinct sentinel for timeout-in-progress state static final Subscription TIMEOUT_MARKER = new Subscription() { ... }; Override // TimerTask.run public void run(Timeout timeout) { if (STATE.compareAndSet(this, STATE_INIT, STATE_TERMINATED)) { // Set S to TIMEOUT_MARKER (not cancelledSubscription!) Subscription current = S.getAndSet(this, TIMEOUT_MARKER); if (current != null) current.cancel(); scheduler.schedule(this); } } Override // Runnable.run - scheduler thread public void run() { // Race 3 detection: if cancel() was called, S is now cancelledSubscription if (s == Operators.cancelledSubscription()) { return; // Cancel won - don't emit anything } // CAS to detect late cancel: TIMEOUT_MARKER → null if (S.compareAndSet(this, TIMEOUT_MARKER, null)) { fallback.subscribe(new FallbackSubscriber<>(actual, this)); } // If CAS failed, cancel() changed S to cancelledSubscription - skip fallback } ``` **Why TIMEOUT_MARKER?** If we used `cancelledSubscription()` in run(Timeout): - `cancel()` calls `Operators.terminate(S, this)` which does: `S.getAndSet(this, cancelledSubscription())` - If S is already `cancelledSubscription()`, this is a no-op! - The scheduler's `run()` would have no way to detect the cancel With TIMEOUT_MARKER: - `cancel()` successfully changes S from TIMEOUT_MARKER → cancelledSubscription() - The scheduler's CAS from TIMEOUT_MARKER → null fails - Cancel is detected, fallback is skipped ### 3. Proper Fallback Handling (Race 2 Fix) The original code subscribed to fallback using method references, which bypassed all cancellation checks. **Before:** ```java private void doTimeoutFallback() { scheduler.schedule(() -> { try { // WRONG: Method references go directly to downstream! // If cancel() is called, these lambdas still execute! fallback.subscribe(actual::onNext, actual::onError, actual::onComplete); } catch (Throwable t) { actual.onError(t); } }); } ``` **Problem:** `actual::onNext` is just a method reference - it has no cancellation check. Even after `cancel()`, fallback signals go directly to downstream. **After:** ```java Override public void run() { // Check cancellation before any signal emission if (s == Operators.cancelledSubscription()) { return; } // Standard fallback path if (S.compareAndSet(this, TIMEOUT_MARKER, null)) { // FallbackSubscriber has defensive checks fallback.subscribe(new FallbackSubscriber<>(actual, this)); } // If CAS failed, cancel() was called - skip fallback } ``` ### 4. FallbackSubscriber with Defensive Checks Even with atomic subscription management, a misbehaving fallback publisher might ignore cancellation and continue emitting signals. FallbackSubscriber adds a final line of defense by checking cancellation state before forwarding each signal. ```java static final class FallbackSubscriber<T> implements CoreSubscriber<T> { Override public void onSubscribe(Subscription s) { // Operators.set() atomically: // 1. Checks if S is already cancelledSubscription() // 2. If so, cancels the incoming subscription and returns false // 3. Otherwise, sets S = s and returns true if (Operators.set(S, parent, s)) { s.request(Long.MAX_VALUE); } // If set() returned false, subscription was rejected } Override public void onNext(T t) { // Defensive check: drop signals if cancelled if (parent.s == Operators.cancelledSubscription()) { Operators.onNextDropped(t, actual.currentContext()); return; } actual.onNext(t); } Override public void onError(Throwable t) { if (parent.s == Operators.cancelledSubscription()) { Operators.onErrorDropped(t, actual.currentContext()); return; } actual.onError(t); } Override public void onComplete() { if (parent.s == Operators.cancelledSubscription()) { return; // Silently drop } actual.onComplete(); } } ``` **Why defensive checks in every method?** A misbehaving publisher might: - Ignore the `cancel()` call on its subscription - Continue emitting values after cancellation - Emit onError or onComplete after cancellation The defensive checks ensure no signals leak to downstream after cancellation. ### 5. Timer Injection for Testability **Before:** ```java // Timer hardcoded - impossible to test race conditions deterministically this.timeout = RpcResources.getHashedWheelTimer().newTimeout(this::doTimeout, delay, unit); ``` **After:** ```java private final Timer timer; // Public constructor uses default timer public MonoTimeoutTransformer(Scheduler scheduler, long delay, TimeUnit unit, Mono<T> fallback) { this(scheduler, delay, unit, fallback, RpcResources.getHashedWheelTimer()); } // Package-private constructor for testing VisibleForTesting MonoTimeoutTransformer(Scheduler scheduler, long delay, TimeUnit unit, Mono<T> fallback, Timer timer) { this.timer = Objects.requireNonNull(timer, "timer"); // ... } ``` ### 6. Input Validation **Before:** ```java public MonoTimeoutTransformer(Scheduler scheduler, long delay, TimeUnit unit, Mono<T> fallback) { this.scheduler = scheduler; // No validation this.delay = delay; // No validation this.unit = unit; // No validation this.fallback = fallback; } ``` **After:** ```java MonoTimeoutTransformer(Scheduler scheduler, long delay, TimeUnit unit, Mono<T> fallback, Timer timer) { this.scheduler = Objects.requireNonNull(scheduler, "scheduler"); if (delay < 0) { throw new IllegalArgumentException("delay must be non-negative, was: " + delay); } this.delay = delay; this.unit = Objects.requireNonNull(unit, "unit"); this.fallback = fallback; // null allowed (means no fallback) this.timer = Objects.requireNonNull(timer, "timer"); } ``` ## Summary of Key Bugs Fixed 1. **Multiple signals possible**: Without atomic state, both source and timeout could emit signals. Now CAS ensures exactly one wins. 2. **Fallback ignores cancellation**: Method references bypassed all checks. Now FallbackSubscriber checks cancellation before every signal. 3. **Cancel after timeout lost**: Using `cancelledSubscription()` as timeout marker made cancel() a no-op. Now TIMEOUT_MARKER is distinct. 4. **Timer not cancelled**: No guarantee timer was cancelled on completion. Now `cancelTimer()` is called in all terminal paths. 5. **Untestable races**: Hardcoded timer made deterministic testing impossible. Now timer is injectable via package-private constructor. 6. **No input validation**: Invalid inputs (null scheduler, negative delay) were silently accepted. Now validated with clear error messages. Reviewed By: RayanRal, adolfojunior Differential Revision: D89081110 fbshipit-source-id: 55a711aef6c80f3ca6b8e6cfb572f29e2c115229

Add ALPN support to thrift java stack Summary: ## What this diff does Introduces a new server transport for Java Thrift — **`UnifiedServerTransport`** — that listens on a single TLS port and uses **ALPN (Application-Layer Protocol Negotiation)** to dynamically select the wire protocol per connection. Two protocols are advertised and supported: `"rs"` (RSocket) and `"thrift"` (THeader/Rocket). This enables in-place migration from Header → Rocket without touching listening ports, DNS, or connection routing. ## Why Today, a Java Thrift server is bound to exactly one wire protocol at process start (`LegacyServerTransportFactory` for Header, `RSocketServerTransportFactory` for Rocket). Migrating a service from Header to Rocket requires a port change, a load-balancer reconfiguration, or a full restart with a new transport. ALPN lets a single server speak both protocols simultaneously, so individual clients can be migrated one at a time without coordinating with the server-side rollout. This matches the C++ ThriftServer behavior, which has supported the same pattern for years. ## Architecture The diff adds five new classes under `com.facebook.thrift.transport.unified`: - **`UnifiedServerTransportFactory`** — `ServerTransportFactory` impl that constructs `UnifiedServerTransport` instances. Plugged into `RpcThriftServer` dispatch (see below). - **`UnifiedServerTransport`** — Reactor-Netty `TcpServer` bound to the configured address with TLS+ALPN. On `doOnConnection`, inspects `SslHandler.applicationProtocol()` and routes: - ALPN selected `"rs"` → adds RSocket length-field decoder, hands the connection to `RSocketServer.asConnectionAcceptor()` wrapped in `TcpDuplexConnection`. - ALPN selected `"thrift"` → adds `ThriftHeaderFrameLengthBasedDecoder` + `LengthFieldPrepender` and hands the connection to `ThriftConnectionAcceptor`. - ALPN absent or unrecognised → **falls back to HEADER** (matches C++ ThriftServer behavior; downgrades from "fail closed" to "fail safe" so legacy clients without ALPN still work). - **`ThriftConnectionAcceptor`** — Reactor-Netty connection handler that bridges Netty inbound bytes to `RpcServerHandler` via `ReactiveHeaderCodec`. Decodes `ByteBuf` → `ThriftFrame`, dispatches request kind (oneway / req-resp), encodes response back, manages timeouts and per-request reference counting. - **`ReactiveHeaderCodec`** — Stateless encoder/decoder for THeader frames, ported from `HeaderTransportCodec` with full parity to landed bug fixes (multi-transform decompression via `CompressionManager` in reverse order, persistent-only header-section parsing, `readString` bounds check, per-header error logging on parse failure). - **`UnifiedServerTransport$ThriftChannelInitializer`** — `ChannelPipelineConfigurer` that enforces `ThriftServerConfig.connectionLimit` and installs `MetricsChannelDuplexHandler` for accepted/rejected/dropped connection counters. Supporting changes: - **`ThriftTransportType`** — adds an `UNKNOWN` enum constant; `fromProtocol(null | unrecognised)` now returns `UNKNOWN` (instead of throwing) so the unified server can apply the fallback-to-HEADER policy uniformly. - **`RpcServerUtils`** — `getSslContext()` configures `ApplicationProtocolConfig` with `[RSOCKET, HEADER]` when `config.isEnableAlpn()`. Also adds `isSslCloseNotify(Throwable)` helper to filter benign TLS shutdown alerts from error logs. - **`RpcThriftServer`** — dispatch logic now routes to `UnifiedServerTransportFactory` first when `config.isEnableAlpn()`, then `RSocketServerTransportFactory` if `useRSocket`, then `LegacyServerTransportFactory` (the old default). - **Client side** (`ThriftClientInitializer`, `LegacyRpcClientFactory`) — propagates the remote `InetSocketAddress` into `SslContext.newHandler(...)` so the client SNI/ALPN handshake completes correctly. No client behavior change when ALPN is disabled. ## Migration strategy **Opt-in via config flag.** The new transport is activated only when `ThriftServerConfig.setEnableAlpn(true)` is called. Default behavior is unchanged — services that don't opt in continue to use `LegacyServerTransportFactory` or `RSocketServerTransportFactory` exactly as before. **Backwards compatible at the wire level.** A unified server with ALPN enabled accepts: - ALPN-aware clients negotiating `"thrift"` → routed to Header path (same wire format as existing Header servers). - ALPN-aware clients negotiating `"rs"` → routed to Rocket path (same wire format as existing Rocket servers). - Legacy clients that don't send an ALPN extension at all → falls back to Header (per the new no-ALPN handling). This means a Header-speaking legacy client can connect to an ALPN-enabled server without modification. **No client SDK changes required for adoption.** Any existing TLS-enabled Java client that does TLS handshake against an ALPN-enabled server will work — either via ALPN negotiation (when the client supports ALPN) or via the no-ALPN HEADER fallback. **Per-service rollout.** Services adopt the unified transport one at a time by setting `setEnableAlpn(true)` in their `ThriftServerConfig`. **Rollback.** A single-line config revert (`setEnableAlpn(false)`) returns a service to `LegacyServerTransportFactory`. No data-format compatibility concerns — the wire formats are unchanged on each ALPN path. Reviewed By: adolfojunior, RayanRal Differential Revision: D87745292 fbshipit-source-id: a5587ceef699ecf6893a9dcd97ab8fab19596af4

Add TLS pipeline for fast thrift server Summary: Adds a TLS pipeline for fast thrift server. Adds a TLS handler to the conneciton pipeline which owns the TLS pipeline. Adds PERMITTED mode for TLS. Reviewed By: robertroeser Differential Revision: D106108957 fbshipit-source-id: a2eed2115a251f60ea83bab304342d6802826d1e

channel_pipeline coro support Summary: Builds on the move-only ContextHandle continuation from previous diff to let a Rust channel_pipeline handler be written as a coroutine instead of a synchronous callback. Adds CallbackContext::spawn, which starts a Send + 'static Rust future on the pipeline's own EventBase. The first poll runs inline in the current callback, so a future that is already ready never leaves the callback frame. If that poll returns Pending, later wakes schedule subsequent polls back onto the same EventBase. The task owns the existing ContextHandle, which retains the pipeline and its EventBase until completion or cancellation, and the completion closure consumes the handle to resume the pipeline from the exact captured position. Adds EventBaseTask (event_base.rs) as the future-to-EventBase bridge, backed by async-task. A BootstrapWake waker services the inline first poll and reconciles it with the async_task Runnable that is only created after that poll returns Pending; wakes arriving before, during, or concurrently with Runnable installation coalesce into exactly one scheduled poll. Both the inline poll and each scheduled poll are wrapped in catch_unwind so a panicking future cannot unwind across the FFI boundary. Extends the FFI surface with FfiCallbackContext::eventBase() and enqueueInEventBase, which moves a raw async_task Runnable through runInEventBaseThreadAlwaysEnqueue. The native side wraps the token in a move-only RAII Task carrying both the call and drop thunks, so an EventBase destroyed with the callback still queued cancels and drops the Rust future exactly once rather than leaking it. Adds CoroReadHandle, CoroWriteHandle, and CoroExceptionHandle adapters plus the ContextReadMessage and ContextWriteMessage traits (implemented for BytesPtr) so a handler body can be an async closure over the message or error. Updates the RustHandler documentation, which previously stated that async work was unsupported. Reviewed By: pranavtbhat Differential Revision: D114866635 fbshipit-source-id: 0efd42a352845e9e647126c60ad0258ef409fe4c

Fix TypeSystem reference model to prevent leaks on recursive types Summary: The existing all-`Arc` model cannot represent recursive types (e.g. `struct Tree { children: list<Tree> }`) without leaking: every inter-node edge holds a strong reference, forming cycles that never reach refcount zero. This switches user-defined edges to `Weak` (while keeping `Arc` for held handles returned by `get()`), so the TypeSystem is the sole strong owner and cycles are broken. Structured fields use deferred initialization via `UnsafeCell<Option<_>>` to allow the builder to create all nodes before resolving cross-references. Also restructures nodes into per-file modules, adds `StructuredNode` trait, `BasicTypeSystem`, and `TypeSystem` trait with `TypeId` resolution. Reviewed By: praihan Differential Revision: D108476964 fbshipit-source-id: 09389ce6be3bed77e9fadf5d329c4938dbeadc2f

context handle for async integration Summary: Adds idiomatic async support to channel_pipeline through two composable, pay-for-use APIs. - Introduces a move-only, one-shot ContextHandle for callback, worker-thread, and other async code. It retains the pipeline across asynchronous work and resumes read, write, or exception processing at the correct handler position on the originating EventBase. - Adds coro::ContextHandle with allocation-free direct awaitables: co_fireRead, co_fireWrite, and co_fireException. Read and write return the actual downstream Result, and the awaiting coroutine resumes on the pipeline EventBase. - Keeps coroutine dependencies isolated from the base pipeline. Existing synchronous handlers and dispatch paths incur no additional runtime cost. - Defines safe behavior when the pipeline closes while work is outstanding: base operations are dropped, while coroutine read and write operations complete with Result::Error. - Adds comprehensive routing, concurrency, ownership, cancellation, destruction, sanitizer, and performance coverage. Reviewed By: pranavtbhat Differential Revision: D113107531 fbshipit-source-id: a4719ca74c5fd9283fe1aa0e7097fe45706ec82a