Replace shim handlers with 5 real per-method server handlers (#725)

Summary:
Pull Request resolved: #725

Programmer-B scope of Phase 6 per `~/feedsim_v2/docs/phase6_researcher_notes.md` section 4. Replaces the Phase 4-B shim handlers in `LeafNodeRank.cc` with five real per-method handlers that mirror the production multifeed_aggregator pipeline (deserialize -> session lookup -> orchestrate on RANKER -> fan out work to GlobalCPUThread / SREventBase -> compress -> sendResponse). Programmer-A's parallel diff (D103795176) added the driver-side `RpcDistRegistry::InboundIdx` enum + accessors with the `kCreateAndPrimeSession` naming chosen here, so the leaf consumes them directly with no further `RpcDistRegistry` changes needed. Programmer-C will delete the legacy `DLRMRequestHandler` / `PageRankRequestHandler` / `AsyncPageRankRequestHandler` and the no-longer-needed thrift/CLI flags in Phase 6/3.

Five changes in `LeafNodeRank.cc`:

1. `SessionState` struct + per-`ThreadData` `folly::F14FastMap<int64_t, SessionState> sessions` map. The driver mints `query_id = (thread_id << 32) | session_counter` so all 4-6 inbound RPCs for one session land on the same RANKER worker — no lock needed, since each `ThreadData` is owned by a single dispatcher thread. `SessionState` carries `query_id`, `user_id`, `created_at_ns`, `session_id`, `mobile_app_version`, plus capped `stream_payloads` / `ifr_payloads` vectors so we pay the prod-equivalent memory cost without unbounded growth.

2. `CreateAndPrimeSessionRequestHandler` — synchronous on `ThriftSrv.IO`. Deserialize the typed request, mint a 32-char hex `session_id` via `makeSessionId(query_id, rng)`, insert into the per-thread `sessions` map, and return a 44 B `CreateAndPrimeSessionResponse`. No DLRM, no fanout, no compression. Latency naturally lands near the prod p50 of 3 ms (`rpc_dist.json`) from the deserialize + map insert; no artificial sleep.

3. `GetStoriesUncompressedRequestHandler` — async. Deserialize on `ThriftSrv.IO`, snapshot `mobile_app_version` into the session, then `folly::via(rankerPool)` to orchestrate. From the orchestrator we (a) `runFeatureExtraction(this_thread, story_contents)` on `GlobalCPUThread` via `folly::via(globalCpu)`, (b) `dlrmInferenceServerSide` on `GlobalCPUThread`, and (c) `issueOutboundFanout(this_thread, args.rpc_fanout_scale_arg)` on `SREventBase` (~94 RPCs at default scale=0.025). All three resolve via `folly::collectAll` before we build the response. New `generateGetStoriesResponse(query_id, num_stories, target_bytes, silesia, rng)` builds a `GetStoriesResponse` with ~100 `RankedStoryInfo`s padded with Silesia bytes so the serialized size hits the rpc_dist.json p50 of 171 KB (or the per-request sample when the inbound section is loaded). Compression runs on `GlobalCPUThread` via `serializeAndCompress`. The wire payload is the uncompressed serialized form (matching the "uncompressed" name of this method); the compressed bytes are computed for cost accounting only.

4. `GetAllStoriesRequestHandler` — async, mirrors handler 3 but with no DLRM or feature extraction (already paid by `getStoriesUncompressed` earlier in the session per researcher §4 row 3) and `issueOutboundFanout(this_thread, args.rpc_fanout_scale_arg * 0.5)` for half-scale fanout. New `generateGetAllStoriesResponse` produces ~900 `RankedStoryInfo`s targeting the rpc_dist.json p50 of 1.47 MB.

5. `StreamDataRequestHandler` — synchronous on `ThriftSrv.IO`. Deserialize, decompress `serialized_payload` (tolerates decompress failures by falling back to the raw bytes — both pay the size cost), append the decompressed bytes to `sessions[query_id].stream_payloads` (capped at 8 entries), and return a 4 B `StreamDataResponse{ack_code=0}`. `StreamIfrPriorityRankingRequestHandler` is async with an analogous shape: `folly::via(rankerPool)` -> parallel decompress on `GlobalCPUThread` + small fanout (`scale * 0.1`) on `SREventBase` -> `collectAll` -> stash bytes -> 4 B ack.

Per-thread thread-pool routing for handler 2 mirrors `~/feedsim_v2/docs/phase6_researcher_notes.md` §4 (and the `multifeed_aggregator` strobelight breakdown):
```
ThriftSrv.IO   --[deserialize, lookup session]-->
RANKER         --[orchestrate]-->
                 ├── GlobalCPUThread [runFeatureExtraction]
                 ├── GlobalCPUThread [DLRM inference]
                 └── SREventBase     [issueOutboundFanout (~94 RPCs)]
                 (await collectAll)
RANKER         --[response struct, serialize]-->
GlobalCPUThread --[compress]-->
ThriftSrv.IO   --[sendResponse]
```

Helper additions (anonymous namespace): `sendThriftResponse<T>(context, response)` (serialize+coalesce+sendResponse), `makeSessionId(query_id, rng)` (32-char hex), `inboundResponseSizeOrDefault(td, idx, fallback)` (samples `RpcDistRegistry::inboundResponseSize(idx)` when loaded, falls back to the prod p50 otherwise — works in OSS builds without an `rpc_dist.json`), `generateGetStoriesResponse` / `generateGetAllStoriesResponse` (build a typed response with Silesia-backed `story_payload` blobs sized to hit a target serialized byte count), and `serializeAndCompress<T>(resp)` (Thrift CompactSerializer + ManagedCompression ZSTD path migrated in Phase 5-C).

Deviation from spec: the `RpcDistRegistry::InboundIdx` enum + `inboundRequestSize` / `inboundResponseSize` / `inboundLatencyUs` accessors that the spec asks Programmer-B to add are already present in the head (D103795176) — Programmer-A added them and explicitly kept the `kCreateAndPrimeSession` naming convention specified for B. No additional `RpcDistRegistry` changes are made by this diff; the leaf consumes the existing accessors from D103795176 directly.

Legacy handlers (`DLRMRequestHandler`, `PageRankRequestHandler`, `AsyncPageRankRequestHandler`) and their `kPageRankRequestType` / `kDLRMRequestType` registrations are left in place per spec — Programmer-C deletes them in Phase 6/3.

Reviewed By: charles-typ

Differential Revision: D103796241

Author: excelle08