feat: add wrap-around and time-based iteration support

Iterator cycling features for workloads exceeding keyspace size:

- Add duration_ms to SamplingConfig for time-based iteration limits
- RandomIter cycles when limit > range_size (e.g., 1M requests on 100K keys)
- SequentialIter auto-enables wrap_around when duration_ms is set
- FilterContext tracks start_time_ms and checks both count and duration limits

New tests for multi-threaded wrap-around behavior:
- test_random_cycling_limit_exceeds_keyspace
- test_random_cycling_multithreaded
- test_random_duration_based_cycling / _multithreaded
- test_sequential_duration_based / _multithreaded
- test_partitioned_with_limit / _multithreaded_disjoint
- test_delete_iter_single_pass
- test_limit_with_duration_limit_wins
- test_duration_with_limit_duration_wins

Documentation updates with examples for:
- Wrap-around mode (.wrap_around())
- Cycling with limit > keyspace
- Time-based iteration (duration_ms)

Author: zvi-code

INTEGRATION_GUIDE.md

@@ -228,16 +228,18 @@ Statistical distributions for realistic workload simulation:
 | **Overlapping** | `.overlapping().random()` | Yes | Contention testing |
 | **Claim** | `.write()` / `.continue_write()` | Yes (atomic) | Exactly-once processing |
 
-### Wrap-Around for Indefinite Iteration
+### Wrap-Around and Time-Based Iteration
 
 By default, iterators stop when the keyspace is exhausted. For benchmarks requiring
-indefinite iteration, use `.wrap_around()`:
+indefinite iteration or fixed-duration runs, use wrap-around or time-based modes:
 
-| Iterator | Default | With `.wrap_around()` |
-|----------|---------|----------------------|
-| `sequential()` | Stops at end | Cycles back to start |
-| `write()` | Stops when full | Clears bitmap, restarts |
-| `random()` | Samples with replacement | Already cycles (no wrap needed) |
+| Iterator | Default | With `.wrap_around()` | With `duration_ms` |
+|----------|---------|----------------------|-------------------|
+| `sequential()` | Stops at end | Cycles back to start | Auto wrap-around |
+| `write()` | Stops when full | Clears bitmap, restarts | N/A |
+| `random()` | Stops at keyspace size | N/A (use limit) | Cycles indefinitely |
+
+#### Wrap-Around Mode
 
 ```rust
 // Sequential read - cycles through existing keys forever
@@ -259,6 +261,51 @@ let handles: Vec<_> = (0..num_workers).map(|_| {
 }).collect();
 ```
 
+#### Random Cycling with Limit > Keyspace
+
+When `limit` exceeds keyspace size, `RandomIter` automatically cycles to fulfill
+the request. This enables "1M requests on 100K keys" scenarios:
+
+```rust
+use keyspace_tracker::SamplingConfig;
+
+// Request 1,000,000 random samples from 100,000-key space
+let items: Vec<_> = tracker.iter()
+    .set_only()
+    .with_sampling(SamplingConfig::default().with_limit(1_000_000))
+    .random()
+    .collect();
+assert_eq!(items.len(), 1_000_000);  // Cycles through keyspace ~10 times
+```
+
+#### Time-Based Iteration
+
+Use `duration_ms` to run for a fixed duration (auto-enables wrap-around):
+
+```rust
+use keyspace_tracker::SamplingConfig;
+
+// Run for 60 seconds, cycling through keyspace
+let items: Vec<_> = tracker.iter()
+    .set_only()
+    .with_sampling(SamplingConfig::default().with_duration_ms(60_000))
+    .random()
+    .collect();
+// Collects items until 60 seconds elapsed
+
+// Combine limit and duration - stops when either is reached first
+let items: Vec<_> = tracker.iter()
+    .set_only()
+    .with_sampling(
+        SamplingConfig::default()
+            .with_limit(1_000_000)
+            .with_duration_ms(30_000)  // 30 seconds max
+    )
+    .random()
+    .collect();
+```
+
+
 ### Partitioning Strategies
 
 | Strategy | Method | Use Case |
@@ -799,6 +846,7 @@ impl<'a> TrackerIterBuilder<'a> {
     pub fn sample(self, probability: f64) -> Self;
     pub fn limit(self, max_items: u64) -> Self;
     pub fn seed(self, seed: u64) -> Self;
+    pub fn with_sampling(self, config: SamplingConfig) -> Self;
 
     // === Distribution ===
     pub fn distribution(self, dist: AccessDistribution) -> Self;
@@ -833,6 +881,49 @@ impl WriteIter<'a> {
 }
 ```
 
+### SamplingConfig
+
+```rust
+/// Configuration for iteration sampling, limits, and duration.
+pub struct SamplingConfig {
+    pub set_ratio: Option<f64>,           // Ratio of set vs unset bits (0.0-1.0)
+    pub limit: Option<u64>,               // Maximum items to return
+    pub duration_ms: Option<u64>,         // Maximum duration in milliseconds
+    pub sample_probability: f64,          // Probabilistic sampling (0.0-1.0)
+    pub seed: Option<u64>,                // Random seed for reproducibility
+    pub distribution: AccessDistribution, // Key access distribution
+    pub overlapping: bool,                // Allow multiple threads to visit same keys
+}
+
+impl SamplingConfig {
+    pub const fn new() -> Self;
+    
+    /// Set mixed ratio: proportion of set (existing) vs unset (new) items.
+    pub const fn with_set_ratio(self, ratio: f64) -> Self;
+    
+    /// Limit number of items returned.
+    /// For RandomIter: if limit > keyspace size, iterator cycles automatically.
+    pub const fn with_limit(self, limit: u64) -> Self;
+    
+    /// Set duration limit in milliseconds.
+    /// Iteration continues (with wrap-around) until duration expires.
+    /// Useful for time-based workloads like "run for 60 seconds".
+    pub const fn with_duration_ms(self, duration_ms: u64) -> Self;
+    
+    /// Set probabilistic sampling (each item has `prob` chance of being returned).
+    pub const fn with_sample_probability(self, prob: f64) -> Self;
+    
+    /// Set seed for reproducible random iteration.
+    pub const fn with_seed(self, seed: u64) -> Self;
+    
+    /// Set access distribution (Uniform, Zipfian, Exponential, etc.).
+    pub const fn with_distribution(self, dist: AccessDistribution) -> Self;
+    
+    /// Enable overlapping mode for contention testing.
+    pub const fn with_overlapping(self, enabled: bool) -> Self;
+}
+```
+
 ### BitmapSnapshot
 
 ```rust

src/config.rs

@@ -68,6 +68,11 @@ pub struct SamplingConfig {
     /// Maximum number of items to return. None = unlimited.
     pub limit: Option<u64>,
 
+    /// Duration limit for iteration (in milliseconds). None = unlimited.
+    /// When set, iteration continues until the duration expires.
+    /// Combines with `limit` - iteration stops when either is reached.
+    pub duration_ms: Option<u64>,
+
     /// Probabilistic sampling ratio (0.0-1.0).
     /// Each matching item has this probability of being returned.
     /// 1.0 = return all matching, 0.5 = return ~50% of matching.
@@ -92,6 +97,7 @@ impl SamplingConfig {
         Self {
             set_ratio: None,
             limit: None,
+            duration_ms: None,
             sample_probability: 1.0,
             seed: None,
             distribution: AccessDistribution::Uniform,
@@ -117,6 +123,16 @@ impl SamplingConfig {
         self
     }
 
+    /// Set duration limit for iteration (in milliseconds).
+    ///
+    /// Iteration will continue (with wraparound) until the duration expires.
+    /// This enables time-based workloads like "run for 60 seconds".
+    #[inline]
+    pub const fn with_duration_ms(mut self, duration_ms: u64) -> Self {
+        self.duration_ms = Some(duration_ms);
+        self
+    }
+
     /// Set probabilistic sampling (each item has `prob` chance of being returned).
     ///
     /// Use this to randomly sample a percentage of the keyspace.
@@ -164,6 +180,7 @@ impl SamplingConfig {
     #[inline]
     pub fn has_sampling(&self) -> bool {
         self.limit.is_some() 
+            || self.duration_ms.is_some()
             || self.sample_probability < 1.0 
             || self.set_ratio.is_some()
             || !matches!(self.distribution, AccessDistribution::Uniform)
@@ -224,28 +241,52 @@ pub struct FilterContext {
     pub(crate) sampling: SamplingConfig,
     rng: fastrand::Rng,
     yielded: u64,
+    /// Start time for duration-based iteration (milliseconds since UNIX epoch).
+    start_time_ms: Option<u64>,
 }
 
 impl FilterContext {
     /// Create a new filter context.
     #[inline]
     pub fn new(filter: BitFilter, sampling: SamplingConfig) -> Self {
         let rng = sampling.make_rng();
+        let start_time_ms = sampling.duration_ms.map(|_| Self::current_time_ms());
         Self {
             filter,
             sampling,
             rng,
             yielded: 0,
+            start_time_ms,
         }
     }
 
-    /// Check if we should continue iterating (respects limit).
+    /// Get current time in milliseconds since UNIX epoch.
+    #[inline]
+    fn current_time_ms() -> u64 {
+        use std::time::{SystemTime, UNIX_EPOCH};
+        SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .map(|d| d.as_millis() as u64)
+            .unwrap_or(0)
+    }
+
+    /// Check if we should continue iterating (respects limit and duration).
     #[inline]
     pub fn check_limit(&self) -> bool {
-        match self.sampling.limit {
-            Some(limit) => self.yielded < limit,
-            None => true,
+        // Check count limit
+        if let Some(limit) = self.sampling.limit {
+            if self.yielded >= limit {
+                return false;
+            }
         }
+        // Check duration limit
+        if let (Some(duration_ms), Some(start_time_ms)) = (self.sampling.duration_ms, self.start_time_ms) {
+            let elapsed = Self::current_time_ms().saturating_sub(start_time_ms);
+            if elapsed >= duration_ms {
+                return false;
+            }
+        }
+        true
     }
 
     /// Check if item matches filter (with mixed-ratio support).