Improve AFL prefix filter fuzzer with clustered keys and ping-pong

Use custom Arbitrary impls for keys and prefixes instead of raw Vec<u8>:
- ClusteredKey: first byte from 0..8, length 1..=9, so keys naturally
  group into a small number of prefix buckets and filter lookups
  frequently hit real data
- ClusteredPrefix: 1..=3 bytes from the same 0..8 alphabet, likely to
  match actual key prefixes
- Key lengths 1..=9 with extractors up to length 4 means AFL naturally
  explores both in-domain and out-of-domain keys

Also restores PrefixPingPong (bidirectional iterator stepping) which
exercises iterator state machine bugs that forward-only scans miss.

Author: zaidoon1

fuzz/prefix_filter/src/main.rs

@@ -59,18 +59,85 @@ impl BpkChoice {
     }
 }
 
+// ---------------------------------------------------------------------------
+// Clustered key/prefix types — small alphabet, bounded length
+// ---------------------------------------------------------------------------
+
+/// A key with first byte drawn from a small alphabet (0..8) and bounded
+/// length (1..=9). This ensures keys cluster into a small number of prefix
+/// groups, so prefix scans and filter lookups frequently hit real data.
+#[derive(Debug, Clone)]
+struct ClusteredKey(Vec<u8>);
+
+impl<'a> Arbitrary<'a> for ClusteredKey {
+    fn arbitrary(u: &mut Unstructured<'a>) -> arbitrary::Result<Self> {
+        let len: usize = u.int_in_range(1..=9)?;
+        let first_byte: u8 = u.int_in_range(0..=7)?;
+        let mut key = Vec::with_capacity(len);
+        key.push(first_byte);
+        for _ in 1..len {
+            key.push(u8::arbitrary(u)?);
+        }
+        Ok(ClusteredKey(key))
+    }
+}
+
+/// A prefix with length 1..=3, each byte from the same small alphabet (0..8).
+/// Likely to match actual key prefixes since keys share the same first-byte space.
+#[derive(Debug, Clone)]
+struct ClusteredPrefix(Vec<u8>);
+
+impl<'a> Arbitrary<'a> for ClusteredPrefix {
+    fn arbitrary(u: &mut Unstructured<'a>) -> arbitrary::Result<Self> {
+        let len: usize = u.int_in_range(1..=3)?;
+        let mut prefix = Vec::with_capacity(len);
+        for _ in 0..len {
+            prefix.push(u.int_in_range(0..=7)?);
+        }
+        Ok(ClusteredPrefix(prefix))
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Operations using clustered keys/prefixes
+// ---------------------------------------------------------------------------
+
 #[derive(Arbitrary, Debug, Clone)]
 enum Op {
-    Insert { key: Vec<u8>, value: Vec<u8> },
-    Delete { key: Vec<u8> },
+    Insert {
+        key: ClusteredKey,
+        value_len: u8,
+        value_seed: u8,
+    },
+    Delete {
+        key: ClusteredKey,
+    },
     Flush,
     Compact,
     MajorCompact,
-    Get { key: Vec<u8> },
-    PrefixScan { prefix: Vec<u8> },
-    PrefixScanRev { prefix: Vec<u8> },
-    RangeScan { start: Vec<u8>, end: Vec<u8> },
-    RangeScanRev { start: Vec<u8>, end: Vec<u8> },
+    Get {
+        key: ClusteredKey,
+    },
+    PrefixScan {
+        prefix: ClusteredPrefix,
+    },
+    PrefixScanRev {
+        prefix: ClusteredPrefix,
+    },
+    /// Bidirectional iterator stepping on a prefix scan.
+    /// Each bool in `directions` controls: true = next_back, false = next.
+    PrefixPingPong {
+        prefix: ClusteredPrefix,
+        directions: Vec<bool>,
+    },
+    RangeScan {
+        start: ClusteredKey,
+        end: ClusteredKey,
+    },
+    RangeScanRev {
+        start: ClusteredKey,
+        end: ClusteredKey,
+    },
     Reopen,
 }
 
@@ -110,6 +177,18 @@ fn open_tree(
     config.open().unwrap()
 }
 
+fn ordered_range<'a>(a: &'a [u8], b: &'a [u8]) -> (&'a [u8], &'a [u8]) {
+    if a <= b {
+        (a, b)
+    } else {
+        (b, a)
+    }
+}
+
+fn make_value(len: u8, seed: u8) -> Vec<u8> {
+    (0..len).map(|i| seed.wrapping_add(i)).collect()
+}
+
 // ---------------------------------------------------------------------------
 // Oracle test: tree with prefix extractor vs tree without
 // ---------------------------------------------------------------------------
@@ -136,11 +215,13 @@ fn run_oracle_test(extractor: SharedPrefixExtractor, bloom_bpk: f32, ops: &[Op])
 
     for (i, op) in ops.iter().enumerate() {
         match op {
-            Op::Insert { key, value } => {
-                // Skip empty keys (lsm-tree requires non-empty keys)
-                if key.is_empty() {
-                    continue;
-                }
+            Op::Insert {
+                key,
+                value_len,
+                value_seed,
+            } => {
+                let key = &key.0;
+                let value = make_value(*value_len, *value_seed);
                 let s1 = seqno_with.next();
                 let s2 = seqno_without.next();
                 tree_with.insert(key.clone(), value.clone(), s1);
@@ -150,9 +231,7 @@ fn run_oracle_test(extractor: SharedPrefixExtractor, bloom_bpk: f32, ops: &[Op])
             }
 
             Op::Delete { key } => {
-                if key.is_empty() {
-                    continue;
-                }
+                let key = &key.0;
                 let s1 = seqno_with.next();
                 let s2 = seqno_without.next();
                 tree_with.remove(key.as_slice(), s1);
@@ -181,9 +260,7 @@ fn run_oracle_test(extractor: SharedPrefixExtractor, bloom_bpk: f32, ops: &[Op])
             }
 
             Op::Get { key } => {
-                if key.is_empty() {
-                    continue;
-                }
+                let key = &key.0;
                 let s1 = vis_with.get();
                 let s2 = vis_without.get();
                 let r1 = tree_with.get(key.as_slice(), s1).unwrap();
@@ -192,6 +269,7 @@ fn run_oracle_test(extractor: SharedPrefixExtractor, bloom_bpk: f32, ops: &[Op])
             }
 
             Op::PrefixScan { prefix } => {
+                let prefix = &prefix.0;
                 let s1 = vis_with.get();
                 let s2 = vis_without.get();
                 let a = collect_kv(tree_with.prefix(prefix.clone(), s1, None));
@@ -200,15 +278,57 @@ fn run_oracle_test(extractor: SharedPrefixExtractor, bloom_bpk: f32, ops: &[Op])
             }
 
             Op::PrefixScanRev { prefix } => {
+                let prefix = &prefix.0;
                 let s1 = vis_with.get();
                 let s2 = vis_without.get();
                 let a = collect_kv(tree_with.prefix(prefix.clone(), s1, None).rev());
                 let b = collect_kv(tree_without.prefix(prefix.clone(), s2, None).rev());
                 assert_eq!(a, b, "op {i}: reverse prefix scan mismatch for {prefix:?}");
             }
 
+            Op::PrefixPingPong { prefix, directions } => {
+                let prefix = &prefix.0;
+                let s1 = vis_with.get();
+                let s2 = vis_without.get();
+                let mut iter_with = tree_with.prefix(prefix.clone(), s1, None);
+                let mut iter_without = tree_without.prefix(prefix.clone(), s2, None);
+
+                for (j, &go_back) in directions.iter().enumerate() {
+                    let item_with = if go_back {
+                        iter_with.next_back()
+                    } else {
+                        iter_with.next()
+                    };
+                    let item_without = if go_back {
+                        iter_without.next_back()
+                    } else {
+                        iter_without.next()
+                    };
+
+                    let kv_with = item_with.and_then(|g| g.into_inner().ok());
+                    let kv_without = item_without.and_then(|g| g.into_inner().ok());
+
+                    match (&kv_with, &kv_without) {
+                        (Some((k1, v1)), Some((k2, v2))) => {
+                            assert_eq!(
+                                (k1.as_ref(), v1.as_ref()),
+                                (k2.as_ref(), v2.as_ref()),
+                                "op {i} step {j}: ping-pong mismatch for prefix {prefix:?}, go_back={go_back}",
+                            );
+                        }
+                        (None, None) => {}
+                        _ => {
+                            panic!(
+                                "op {i} step {j}: ping-pong length mismatch for prefix {prefix:?}: \
+                                 with={kv_with:?}, without={kv_without:?}",
+                            );
+                        }
+                    }
+                }
+            }
+
             Op::RangeScan { start, end } => {
-                let (lo, hi) = ordered_range(start, end);
+                let (lo, hi) = ordered_range(&start.0, &end.0);
                 if lo == hi {
                     continue;
                 }
@@ -220,7 +340,7 @@ fn run_oracle_test(extractor: SharedPrefixExtractor, bloom_bpk: f32, ops: &[Op])
             }
 
             Op::RangeScanRev { start, end } => {
-                let (lo, hi) = ordered_range(start, end);
+                let (lo, hi) = ordered_range(&start.0, &end.0);
                 if lo == hi {
                     continue;
                 }
@@ -251,14 +371,6 @@ fn run_oracle_test(extractor: SharedPrefixExtractor, bloom_bpk: f32, ops: &[Op])
     }
 }
 
-fn ordered_range<'a>(a: &'a [u8], b: &'a [u8]) -> (&'a [u8], &'a [u8]) {
-    if a <= b {
-        (a, b)
-    } else {
-        (b, a)
-    }
-}
-
 // ---------------------------------------------------------------------------
 // AFL entry point
 // ---------------------------------------------------------------------------
@@ -270,10 +382,18 @@ fn main() {
             return;
         };
 
-        // Limit op count so each iteration stays fast for AFL
+        // Limit op count so each iteration stays fast for AFL.
+        // Also cap PrefixPingPong directions to avoid very slow iterations.
         if input.ops.is_empty() || input.ops.len() > 200 {
             return;
         }
+        for op in &input.ops {
+            if let Op::PrefixPingPong { directions, .. } = op {
+                if directions.len() > 50 {
+                    return;
+                }
+            }
+        }
 
         run_oracle_test(
             input.extractor.into_extractor(),