Migrate compression to ManagedCompression (#722)

Summary:
Pull Request resolved: #722

Migrate the three ZSTD compression callsites in `LeafNodeRank.cc` from raw `folly::compression::getCodec(CodecType::ZSTD)` to ManagedCompression, the documented Meta standard for application-level compression in fbcode (per `fbcode/.llms/rules/managed_compression.md` and the `managed_compression_integration` skill). ManagedCompression handles dictionary training, parameter tuning, and rollout via infrastructure rather than hard-coded codec choice/level.

Three callsites migrated, two categories:
- `compressPayload` (line ~399) — `leaf_random_string` category
- `decompressPayload` (line ~409) — `leaf_random_string` category (same as compressPayload, since both operate on the same pseudo-random payload bytes — required so ManagedCompression serves the right dictionary on decompress)
- `compressThrift` (line ~416) — `leaf_thrift_payload` category (serialized RankingResponse / CompactProtocol — distinct payload shape)

Following the canonical pattern from `common/managed_compression/examples/ManagedCompressionExample.cpp`:
- One process-wide `folly::Singleton<ManagedCompressionFactory>` keyed by oncall=`chips_dcperf` and project=`feedsim`. Constructing a factory per call is explicitly discouraged.
- `getCachedCodec(category)` per category, preferred over `getCodec()` for hot paths.
- 2 categories, both clearly distinct payload shapes (random bytes vs thrift CompactProtocol). Per the skill, category count is kept modest.

Open-source build path: the benchpress repo is open-sourced and ManagedCompression is internal-only, so the include and use sites are gated behind `#ifdef BENCHPRESS_INTERNAL`. When the gate is undefined (the current OSS / CMake build path that all install scripts use today), the original raw folly ZSTD code remains as the fallback so the OSS build still works. `LeafNodeRank.cc` has no Buck build target — it is built only by CMake at fbpkg-install time — so no Buck-side wiring is needed in this commit. A follow-up (Phase 6) can add `-DBENCHPRESS_INTERNAL=1` to the internal CMake invocation to flip the gate on.

Stack position: depends on `bebd655f6d` (Phase 4-B thrift structs). Sibling of Phase 5-A mock_services (`bd2517aa83`).

Reviewed By: charles-typ

Differential Revision: D103768051

fbshipit-source-id: 307a9a1e9d939a7feeb0a673997824134eaf792c

Author: excelle08

packages/feedsim/third_party/src/workloads/ranking/LeafNodeRank.cc

@@ -28,6 +28,21 @@
 #include <folly/Range.h>
 #include <folly/compression/Compression.h>
 #include <folly/executors/CPUThreadPoolExecutor.h>
+
+// ManagedCompression is the documented Meta standard for application-level
+// compression (per fbcode/.llms/rules/managed_compression.md), but it is
+// internal-only. The benchpress repo is open-sourced, so the include and
+// usages below are gated behind BENCHPRESS_INTERNAL. When the gate is not
+// defined (the current OSS / CMake build path) we fall back to raw folly
+// ZSTD, which is the historical behavior. The gate is wired up by the
+// fbcode-internal Buck build (see packages/feedsim/.../mock_services/BUCK
+// for the analogous internal-only wiring); the open-source CMake build
+// leaves it undefined.
+#ifdef BENCHPRESS_INTERNAL
+#include <folly/Singleton.h>
+#include "common/managed_compression/ManagedCompression.h"
+#endif
+
 #include <folly/executors/GlobalExecutor.h>
 #include <folly/executors/IOThreadPoolExecutor.h>
 #include <folly/futures/Future.h>
@@ -396,29 +411,84 @@ void ThreadStartup(
   }
 }
 
-std::string compressPayload(const std::string& data, int result) {
+#ifdef BENCHPRESS_INTERNAL
+namespace {
+using facebook::managed_compression::ManagedCompressionFactory;
+
+// One ManagedCompressionFactory per (oncall, project) pair, lifetime =
+// process. Per the ManagedCompression skill / wiki, constructing a new
+// factory per call is expensive and explicitly discouraged.
+//
+// Two categories are used in this file:
+//   "leaf_random_string"   — the pseudo-random payload bytes shared by
+//                            compressPayload / decompressPayload. Both
+//                            sides MUST use the same category so
+//                            ManagedCompression can serve the right
+//                            dictionary on decompress.
+//   "leaf_thrift_payload"  — serialized RankingResponse (CompactProtocol)
+//                            consumed by compressThrift.
+class FeedSimCompressionTag {};
+folly::Singleton<ManagedCompressionFactory, FeedSimCompressionTag> gFactory(
+    [] {
+      return new ManagedCompressionFactory(
+          /*oncall_team=*/"chips_dcperf",
+          /*project=*/"feedsim");
+    });
+
+std::shared_ptr<folly::compression::Codec> getRandomStringCodec() {
+  // getCachedCodec() reuses the codec instance per category; preferred
+  // over getCodec() for hot paths per references/cpp.md. try_get() can
+  // return null before SingletonVault::registrationComplete() or after
+  // destroyInstances() during shutdown — fail loudly rather than crash
+  // with a null-deref.
+  auto factory = gFactory.try_get();
+  CHECK(factory) << "ManagedCompressionFactory singleton unavailable";
+  return factory->getCachedCodec("leaf_random_string");
+}
+
+std::shared_ptr<folly::compression::Codec> getThriftPayloadCodec() {
+  auto factory = gFactory.try_get();
+  CHECK(factory) << "ManagedCompressionFactory singleton unavailable";
+  return factory->getCachedCodec("leaf_thrift_payload");
+}
+} // namespace
+#endif // BENCHPRESS_INTERNAL
+
+std::string compressPayload(const std::string& data, int /*result*/) {
   folly::StringPiece output(
       data.data(),
       std::min(args.compression_data_size_arg, args.random_data_size_arg));
+#ifdef BENCHPRESS_INTERNAL
+  return getRandomStringCodec()->compress(output);
+#else
   auto codec =
       folly::compression::getCodec(folly::compression::CodecType::ZSTD);
   std::string compressed = codec->compress(output);
   return std::move(compressed);
+#endif
 }
 
 std::string decompressPayload(const std::string& data) {
+#ifdef BENCHPRESS_INTERNAL
+  return getRandomStringCodec()->uncompress(data);
+#else
   auto codec =
       folly::compression::getCodec(folly::compression::CodecType::ZSTD);
   std::string decompressed = codec->uncompress(data);
   return decompressed;
+#endif
 }
 
 std::unique_ptr<folly::IOBuf> compressThrift(
     std::unique_ptr<folly::IOBuf> buf) {
+#ifdef BENCHPRESS_INTERNAL
+  return getThriftPayloadCodec()->compress(buf.get());
+#else
   auto codec =
       folly::compression::getCodec(folly::compression::CodecType::ZSTD);
   auto compressed_buf = codec->compress(buf.get());
   return compressed_buf;
+#endif
 }
 
 folly::IOBufQueue serializePayload(const ranking::RankingResponse& resp) {