Validate remote compaction snapshot ordering (#15160)

Summary:
Remote compaction passes snapshot sequence numbers through serialized `CompactionServiceInput`. Downstream compaction logic assumes the sequence numbers are strictly increasing when selecting the earliest snapshot and using `std::lower_bound()` for snapshot visibility. Previously, malformed input could bypass the debug-only ordering assertion in release builds and select an incorrect visibility boundary.

- Validate decoded snapshot sequence numbers before remote compaction setup and return `InvalidArgument` for duplicate or descending values.
- Document the strict ordering contract while retaining the `CompactionIterator` assertion as defense in depth for internally constructed inputs.
- Canonicalize randomized serialization-test snapshots and add coverage for malformed snapshot lists through `DB::OpenAndCompact()`.

Pull Request resolved: https://github.com/facebook/rocksdb/pull/15160

Test Plan:
- Built `compaction_job_test` and ran `CompactionJobTest.OpenAndCompactRejectsUnorderedSnapshots` in the normal configuration.
- Ran the focused test with `ASSERT_STATUS_CHECKED=1`.
- Ran the focused test for 100 repetitions.

Reviewed By: rban1

Differential Revision: D117888826

Pulled By: xingbowang

fbshipit-source-id: 2312205acfe17271c2083808d372c93a5914114c

db/compaction/compaction_job.h

@@ -591,6 +591,7 @@ class CompactionJob {
 struct CompactionServiceInput {
   std::string cf_name;
 
+  // Snapshot sequence numbers in strictly increasing order.
   std::vector<SequenceNumber> snapshots;
 
   // SST files for compaction, it should already be expended to include all the

db/compaction/compaction_job_test.cc

@@ -1572,6 +1572,10 @@ TEST_F(CompactionJobTest, InputSerialization) {
   while (!rnd.OneIn(10)) {
     input.snapshots.emplace_back(rnd64.Uniform(UINT64_MAX));
   }
+  std::sort(input.snapshots.begin(), input.snapshots.end());
+  input.snapshots.erase(
+      std::unique(input.snapshots.begin(), input.snapshots.end()),
+      input.snapshots.end());
   while (!rnd.OneIn(10)) {
     input.input_files.emplace_back(rnd.RandomString(
         rnd.Uniform(kStrMaxLen - 1) +
@@ -1641,6 +1645,24 @@ TEST_F(CompactionJobTest, InputSerialization) {
   ASSERT_TRUE(s.IsNotSupported());
 }
 
+TEST_F(CompactionJobTest, OpenAndCompactRejectsUnorderedSnapshots) {
+  for (const std::vector<SequenceNumber>& snapshots :
+       {std::vector<SequenceNumber>{2, 1}, std::vector<SequenceNumber>{1, 1}}) {
+    CompactionServiceInput input;
+    input.snapshots = snapshots;
+
+    std::string serialized_input;
+    ASSERT_OK(input.Write(&serialized_input));
+
+    std::string serialized_result;
+    Status s = DB::OpenAndCompact("", "", serialized_input, &serialized_result,
+                                  CompactionServiceOptionsOverride());
+    ASSERT_TRUE(s.IsInvalidArgument());
+    ASSERT_NE(s.ToString().find("snapshots must be strictly increasing"),
+              std::string::npos);
+  }
+}
+
 TEST_F(CompactionJobTest, ResultSerialization) {
   // Setup a random CompactionServiceResult
   CompactionServiceResult result;

db/compaction/compaction_service_job.cc

@@ -1085,9 +1085,19 @@ Status CompactionServiceInput::Read(const std::string& data_str,
     ConfigOptions cf;
     cf.invoke_prepare_options = false;
     cf.ignore_unknown_options = true;
-    return OptionTypeInfo::ParseType(
+    Status s = OptionTypeInfo::ParseType(
         cf, data_str.substr(sizeof(BinaryFormatVersion)), cs_input_type_info,
         obj);
+    if (!s.ok()) {
+      return s;
+    }
+    for (size_t i = 1; i < obj->snapshots.size(); ++i) {
+      if (obj->snapshots[i - 1] >= obj->snapshots[i]) {
+        return Status::InvalidArgument(
+            "CompactionServiceInput snapshots must be strictly increasing");
+      }
+    }
+    return Status::OK();
   } else {
     return Status::NotSupported(
         "Compaction Service Input data version not supported: " +