dyniak 2.0.0

Riak-compatible protocol surface (HTTP + PBC) and storage bridge for the Dynomite Rust port
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
//! Walk-N-successors replication planning for Riak buckets.
//!
//! The default Dynomite replication strategy
//! ([`ReplicationStrategy::Topology`]) fans a write out across
//! datacenters and racks per the configured consistency level.
//! Riak instead replicates a key to the primary partition plus
//! the next `n_val - 1` peers reached by walking forward on the
//! token ring, skipping a peer that has already been added so a
//! peer with multiple ring slots cannot count more than once.
//!
//! This module provides the [`plan_replicas`] entry point and
//! the supporting [`RingView`] accessor. The dispatcher consults
//! a per-bucket-property
//! [`crate::proto::pb::RpbBucketProps::replication_strategy`]
//! field to decide which path to take. When the field is unset
//! (or set to [`REPLICATION_STRATEGY_TOPOLOGY`]), the existing
//! topology code stays in charge; when it is
//! [`REPLICATION_STRATEGY_SUCCESSORS`], this module returns the
//! replica list.
//!
//! [`REPLICATION_STRATEGY_TOPOLOGY`]: crate::proto::pb::REPLICATION_STRATEGY_TOPOLOGY
//! [`REPLICATION_STRATEGY_SUCCESSORS`]: crate::proto::pb::REPLICATION_STRATEGY_SUCCESSORS
//!
//! # Edge cases
//!
//! * Fewer peers than `n_val` -- the plan returns whatever peers
//!   are available; a `tracing::warn!` is emitted at config
//!   validation time so operators see the under-provision early.
//! * Down peers -- the no-liveness [`plan_replicas`] entry point
//!   does NOT skip them during planning; they are returned as
//!   targets so the runtime
//!   [`dynomite::cluster::peer::PeerState`]-aware filter in the
//!   dispatcher decides whether to send (this matches the
//!   topology mode's behaviour). [`plan_replicas_with_liveness`]
//!   is the liveness-aware variant: a primary-window peer it is
//!   told is down is dropped from the plan and a successor node
//!   past the `n_val` window stands in for it, marked
//!   [`ReplicaTarget::is_fallback`] (Riak's sloppy quorum /
//!   hinted-handoff preference-list substitution).
//!
//! # Examples
//!
//! ```
//! use dyniak::replication::{plan_replicas, ReplicationPlan, ReplicationStrategy, RingPoint, RingView};
//! use dynomite::msg::ConsistencyLevel;
//!
//! // 5 peers at evenly spaced tokens.
//! let span = u64::from(u32::MAX);
//! let points = (0..5u32)
//!     .map(|i| RingPoint::new(u64::from(i) * span / 5, i, "dc1", "r1"))
//!     .collect();
//! let ring = RingView::new(points);
//!
//! // Key hashes to peer 1's slot (next slot above token 0).
//! let plan = plan_replicas(&ring, 1, 3, ReplicationStrategy::Successors, ConsistencyLevel::DcOne);
//! match plan {
//!     ReplicationPlan::Successors { successors, primary, .. } => {
//!         let mut idxs = vec![primary.peer_idx];
//!         idxs.extend(successors.iter().map(|r| r.peer_idx));
//!         assert_eq!(idxs, vec![1, 2, 3]);
//!     }
//!     other @ ReplicationPlan::Topology(_) => panic!("expected successors plan, got {other:?}"),
//! }
//! ```

use dynomite::cluster::ReplicaTarget;
use dynomite::msg::ConsistencyLevel;

use crate::proto::pb::{REPLICATION_STRATEGY_SUCCESSORS, REPLICATION_STRATEGY_TOPOLOGY};

/// Strategy used to choose the replica target list for a key.
///
/// Riak-mode pools default to [`Self::Successors`]; non-Riak
/// pools default to [`Self::Topology`] and never expose this
/// knob. The wire encoding lives at
/// [`crate::proto::pb::RpbBucketProps::replication_strategy`].
#[derive(Copy, Clone, Debug, Default, Eq, PartialEq)]
pub enum ReplicationStrategy {
    /// Dynomite's classic per-DC, per-rack quorum fan-out. The
    /// dispatcher uses the existing
    /// [`dynomite::cluster::dispatch::ClusterDispatcher::plan`]
    /// pipeline.
    #[default]
    Topology,
    /// Riak-style walk-N-successors. The primary peer plus the
    /// next `n_val - 1` peers reached by walking forward on the
    /// ring, deduplicated.
    Successors,
}

/// Errors produced when decoding a wire `replication_strategy`
/// selector.
#[derive(Debug, thiserror::Error, Eq, PartialEq)]
#[non_exhaustive]
pub enum ReplicationStrategyError {
    /// The wire value was outside the documented enum range.
    #[error("replication_strategy: unknown selector {0}")]
    Unknown(u32),
}

impl ReplicationStrategy {
    /// Convert a wire selector to the in-memory enum. `None`
    /// means the field is unset on the wire and the caller's
    /// default applies (mode-aware: Riak buckets default to
    /// Successors).
    ///
    /// # Errors
    ///
    /// Returns [`ReplicationStrategyError::Unknown`] when the
    /// wire value is outside the documented enum range.
    pub fn from_wire(value: u32) -> Result<Self, ReplicationStrategyError> {
        match value {
            REPLICATION_STRATEGY_TOPOLOGY => Ok(Self::Topology),
            REPLICATION_STRATEGY_SUCCESSORS => Ok(Self::Successors),
            other => Err(ReplicationStrategyError::Unknown(other)),
        }
    }

    /// Convert the in-memory enum to the wire selector.
    #[must_use]
    pub fn to_wire(self) -> u32 {
        match self {
            Self::Topology => REPLICATION_STRATEGY_TOPOLOGY,
            Self::Successors => REPLICATION_STRATEGY_SUCCESSORS,
        }
    }
}

/// One ring entry: a `(token, peer_idx)` mapping plus the
/// peer's `(dc, rack)` labels so the planner can build a
/// [`ReplicaTarget`] without a second pool look-up.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct RingPoint {
    /// Token at this ring position.
    pub token: u64,
    /// Index into the pool's peer array.
    pub peer_idx: u32,
    /// Datacenter name.
    pub dc: String,
    /// Rack name.
    pub rack: String,
}

impl RingPoint {
    /// Build a ring point. Convenience constructor for tests and
    /// integrations that have ASCII-friendly DC / rack names.
    pub fn new(token: u64, peer_idx: u32, dc: impl Into<String>, rack: impl Into<String>) -> Self {
        Self {
            token,
            peer_idx,
            dc: dc.into(),
            rack: rack.into(),
        }
    }
}

/// Read-only view onto the cluster ring used by
/// [`plan_replicas`]. Holds the ordered list of `(token,
/// peer_idx)` pairs across all datacenters and racks. The
/// dispatcher builds one of these from the live
/// [`dynomite::cluster::ServerPool`] when it needs to invoke the
/// successors strategy; tests construct one directly from a
/// fixture vector.
#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct RingView {
    points: Vec<RingPoint>,
}

impl RingView {
    /// Wrap a set of [`RingPoint`]s in a ring view, sorted by
    /// token ascending so the walk produces a deterministic
    /// result.
    ///
    /// # Examples
    ///
    /// ```
    /// use dyniak::replication::{RingPoint, RingView};
    /// let pts = vec![
    ///     RingPoint::new(20, 1, "dc1", "r1"),
    ///     RingPoint::new(10, 0, "dc1", "r1"),
    /// ];
    /// let ring = RingView::new(pts);
    /// assert_eq!(ring.points().len(), 2);
    /// assert_eq!(ring.points()[0].peer_idx, 0);
    /// ```
    #[must_use]
    pub fn new(mut points: Vec<RingPoint>) -> Self {
        points.sort_by_key(|p| p.token);
        Self { points }
    }

    /// Borrow the ordered ring points.
    #[must_use]
    pub fn points(&self) -> &[RingPoint] {
        &self.points
    }

    /// Count of distinct peer indices in the ring.
    #[must_use]
    pub fn peer_count(&self) -> usize {
        let mut idxs: Vec<u32> = self.points.iter().map(|p| p.peer_idx).collect();
        idxs.sort_unstable();
        idxs.dedup();
        idxs.len()
    }

    /// Find the index into [`Self::points`] that owns
    /// `key_hash`: the smallest entry whose token is greater
    /// than or equal to `key_hash`, wrapping to entry 0 when
    /// the hash is greater than every token.
    ///
    /// Returns `None` when the ring is empty.
    #[must_use]
    pub fn primary_index(&self, key_hash: u64) -> Option<usize> {
        if self.points.is_empty() {
            return None;
        }
        match self.points.binary_search_by_key(&key_hash, |p| p.token) {
            Ok(i) => Some(i),
            Err(i) => {
                if i >= self.points.len() {
                    Some(0)
                } else {
                    Some(i)
                }
            }
        }
    }
}

/// Plan produced by [`plan_replicas`].
///
/// `Topology` carries the targets the existing topology pipeline
/// produced; the variant exists so a caller can return a uniform
/// type from both branches. `Successors` carries the primary
/// plus the walk-forward successors; `n` echoes the requested
/// `n_val` so callers can spot-check under-provisioned rings.
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum ReplicationPlan {
    /// Topology fan-out targets. The dispatcher's existing
    /// pipeline is the source of truth; this variant is the
    /// pass-through container [`plan_replicas`] returns when
    /// the strategy is [`ReplicationStrategy::Topology`].
    Topology(Vec<ReplicaTarget>),
    /// Walk-N-successors replica list.
    Successors {
        /// Primary peer for the key.
        primary: ReplicaTarget,
        /// Requested replication factor (`n_val`). May be larger
        /// than `successors.len() + 1` when the ring has fewer
        /// distinct peers than `n_val`.
        n: u8,
        /// `n - 1` (or fewer) successor peers, in walk order.
        /// Excludes the primary; the primary is carried in
        /// `Self::Successors::primary`.
        successors: Vec<ReplicaTarget>,
    },
}

impl ReplicationPlan {
    /// Flatten the plan into a single replica list (primary
    /// first, then successors, in walk order). Convenience for
    /// callers that hand the targets straight to the per-peer
    /// outbound channels.
    #[must_use]
    pub fn into_replica_list(self) -> Vec<ReplicaTarget> {
        match self {
            Self::Topology(targets) => targets,
            Self::Successors {
                primary,
                mut successors,
                ..
            } => {
                let mut out = Vec::with_capacity(1 + successors.len());
                out.push(primary);
                out.append(&mut successors);
                out
            }
        }
    }
}

/// Compute the replica plan for a key according to `strategy`.
///
/// `consistency` is propagated by the dispatcher into the
/// returned plan; this function does not consult it (the
/// successors walk is consistency-level agnostic), but the
/// signature carries it so a future strategy that does care has
/// the value at hand.
///
/// # Behaviour
///
/// * [`ReplicationStrategy::Topology`] -- returns
///   `ReplicationPlan::Topology(Vec::new())`. The caller is
///   expected to fall back to the existing topology pipeline
///   for the actual targets; the empty vector is the sentinel.
/// * [`ReplicationStrategy::Successors`] -- walks the ring
///   forward from the primary slot, deduplicating peer indices,
///   and returns up to `n_val` distinct peers. When the ring
///   has fewer peers than `n_val`, the result has
///   `successors.len() + 1 < n_val`; the caller is expected to
///   surface the under-provision via `tracing::warn!` at config
///   validation time.
///
/// Returns `ReplicationPlan::Successors` with an empty
/// `successors` vector when the ring has exactly one distinct
/// peer; `n_val == 0` is treated identically to `n_val == 1`
/// for safety so the dispatcher always has a primary.
///
/// # Examples
///
/// See the module-level example.
#[must_use]
pub fn plan_replicas(
    distribution: &RingView,
    key_hash: u64,
    n_val: u8,
    strategy: ReplicationStrategy,
    consistency: ConsistencyLevel,
) -> ReplicationPlan {
    let _ = consistency;
    if matches!(strategy, ReplicationStrategy::Topology) {
        return ReplicationPlan::Topology(Vec::new());
    }
    plan_successors(distribution, key_hash, n_val)
}

fn plan_successors(distribution: &RingView, key_hash: u64, n_val: u8) -> ReplicationPlan {
    let target_count = (n_val as usize).max(1);
    let points = distribution.points();
    let Some(start) = distribution.primary_index(key_hash) else {
        // Empty ring: synthesise a placeholder primary so the
        // dispatcher's downstream code never has to handle the
        // "no primary" branch under successors mode. The local
        // node's `is_routable()` filter will subsequently reject
        // it.
        return ReplicationPlan::Successors {
            primary: ReplicaTarget {
                peer_idx: 0,
                dc: String::new(),
                rack: String::new(),
                is_local: false,
                is_fallback: false,
            },
            n: n_val,
            successors: Vec::new(),
        };
    };
    let mut chosen: Vec<ReplicaTarget> = Vec::with_capacity(target_count);
    let len = points.len();
    for step in 0..len {
        if chosen.len() >= target_count {
            break;
        }
        let idx = (start + step) % len;
        let pt = &points[idx];
        if chosen.iter().any(|t| t.peer_idx == pt.peer_idx) {
            continue;
        }
        chosen.push(ReplicaTarget {
            peer_idx: pt.peer_idx,
            dc: pt.dc.clone(),
            rack: pt.rack.clone(),
            is_local: false,
            // Walking the ring never emits more than `target_count`
            // (== n_val) entries, so every entry `plan_successors`
            // produces is one of the n_val primary owners. Nothing in
            // this planner's inputs (a `RingView`, which carries no
            // liveness) can identify a down primary and substitute a
            // fallback, so `is_fallback` is always `false` here. The
            // real fallback-substitution planner is
            // [`plan_successors_with_liveness`] below, which the
            // dispatcher uses once a `RoutingHooks` liveness source is
            // wired; see its doc comment for the counting contract PR
            // relies on.
            is_fallback: false,
        });
    }
    let mut iter = chosen.into_iter();
    let primary = iter
        .next()
        .expect("primary_index returned Some so len >= 1");
    let successors: Vec<ReplicaTarget> = iter.collect();
    ReplicationPlan::Successors {
        primary,
        n: n_val,
        successors,
    }
}

/// Liveness oracle consulted by [`plan_replicas_with_liveness`].
///
/// Kept minimal (one method, peer index in, bool out) so a caller can
/// implement it against whatever liveness source it has -- today the
/// production wiring has none (see [`plan_replicas`]'s doc comment),
/// but the type exists so the primary-vs-fallback counting path this
/// module and [`crate::router::RouteDecision`] build on is exercised
/// by a real implementation in tests, not asserted by inspection.
pub trait ReplicaLiveness: Send + Sync + std::fmt::Debug {
    /// True when `peer_idx` is known reachable. A liveness source
    /// that has no opinion about a peer should return `true`: absence
    /// of information is not evidence of failure, and treating an
    /// unknown peer as down would substitute a fallback for a peer
    /// that may in fact be fine.
    fn is_up(&self, peer_idx: u32) -> bool;
}

/// Walk-N-successors planning with sloppy-quorum fallback
/// substitution.
///
/// Identical to [`plan_replicas`] / `plan_successors` except that a
/// primary-window peer (one of the first `n_val` distinct peers
/// reached walking forward from the key's slot) that `liveness`
/// reports down is dropped, and the walk continues past the `n_val`
/// window to find a stand-in. The stand-in is marked
/// [`ReplicaTarget::is_fallback`] so [`crate::router::RouteDecision`]
/// and the request handlers can count primary responses separately
/// from fallback responses, which is what the PR / PW quorum knobs
/// need to be enforced correctly.
///
/// When every peer up to the ring's full distinct-peer count is down,
/// the returned plan may carry fewer than `n_val` targets (there is
/// nothing left to hand out); callers see this as
/// `replica_list().len() < n_val`, exactly as the no-liveness planner
/// already tolerates an under-provisioned ring.
///
/// # Examples
///
/// ```
/// use dyniak::replication::{
///     plan_replicas_with_liveness, ReplicaLiveness, ReplicationPlan, RingPoint, RingView,
/// };
///
/// #[derive(Debug)]
/// struct DownPeers(Vec<u32>);
/// impl ReplicaLiveness for DownPeers {
///     fn is_up(&self, peer_idx: u32) -> bool {
///         !self.0.contains(&peer_idx)
///     }
/// }
///
/// let span = u64::from(u32::MAX);
/// let points = (0..5u32)
///     .map(|i| RingPoint::new(u64::from(i) * span / 5, i, "dc1", "r1"))
///     .collect();
/// let ring = RingView::new(points);
///
/// // n_val=3 from peer 1's slot names peers [1, 2, 3]; peer 2 is down,
/// // so the walk continues to peer 4 as its fallback stand-in.
/// let plan = plan_replicas_with_liveness(&ring, 1, 3, &DownPeers(vec![2]));
/// let ReplicationPlan::Successors { primary, successors, .. } = plan else {
///     panic!("expected successors plan");
/// };
/// assert_eq!(primary.peer_idx, 1);
/// assert!(!primary.is_fallback);
/// let idxs: Vec<(u32, bool)> = successors.iter().map(|t| (t.peer_idx, t.is_fallback)).collect();
/// assert_eq!(idxs, vec![(3, false), (4, true)]);
/// ```
#[must_use]
pub fn plan_replicas_with_liveness(
    distribution: &RingView,
    key_hash: u64,
    n_val: u8,
    liveness: &dyn ReplicaLiveness,
) -> ReplicationPlan {
    plan_successors_with_liveness(distribution, key_hash, n_val, liveness)
}

fn plan_successors_with_liveness(
    distribution: &RingView,
    key_hash: u64,
    n_val: u8,
    liveness: &dyn ReplicaLiveness,
) -> ReplicationPlan {
    let target_count = (n_val as usize).max(1);
    let points = distribution.points();
    let Some(start) = distribution.primary_index(key_hash) else {
        return ReplicationPlan::Successors {
            primary: ReplicaTarget {
                peer_idx: 0,
                dc: String::new(),
                rack: String::new(),
                is_local: false,
                is_fallback: false,
            },
            n: n_val,
            successors: Vec::new(),
        };
    };
    let len = points.len();
    // Single pass over the ring's distinct peers: the first
    // `target_count` LIVE peers walked fill the primary window; a
    // down peer within the window is dropped (not carried forward),
    // and every distinct peer walked once the window is full is a
    // fallback candidate. This is exactly Riak's preference-list
    // construction: primaries first, then however many fallback
    // (hinted-handoff) nodes are needed to backfill the ones that were
    // down.
    let mut primaries: Vec<ReplicaTarget> = Vec::with_capacity(target_count);
    let mut fallback_pool: Vec<ReplicaTarget> = Vec::new();
    let mut seen: Vec<u32> = Vec::new();
    for step in 0..len {
        let idx = (start + step) % len;
        let pt = &points[idx];
        if seen.contains(&pt.peer_idx) {
            continue;
        }
        seen.push(pt.peer_idx);
        let target = ReplicaTarget {
            peer_idx: pt.peer_idx,
            dc: pt.dc.clone(),
            rack: pt.rack.clone(),
            is_local: false,
            is_fallback: false,
        };
        if seen.len() <= target_count {
            if liveness.is_up(pt.peer_idx) {
                primaries.push(target);
            }
        } else {
            fallback_pool.push(target);
        }
    }
    while primaries.len() < target_count {
        let Some(pos) = fallback_pool
            .iter()
            .position(|t| liveness.is_up(t.peer_idx))
        else {
            // Every peer beyond the window is also down (or the ring
            // has fewer distinct peers than `n_val`): the plan is
            // under-provisioned, matching the no-liveness planner's
            // documented tolerance for a short ring.
            break;
        };
        let mut stand_in = fallback_pool.remove(pos);
        stand_in.is_fallback = true;
        primaries.push(stand_in);
    }
    let mut iter = primaries.into_iter();
    let primary = iter
        .next()
        .expect("primary_index returned Some so len >= 1");
    let successors: Vec<ReplicaTarget> = iter.collect();
    ReplicationPlan::Successors {
        primary,
        n: n_val,
        successors,
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn five_peer_ring() -> RingView {
        // 5 peers at tokens 0/0.4G/0.8G/1.2G/1.6G of u32::MAX
        // (rounded to integer steps).
        let span = u64::from(u32::MAX);
        let mut pts: Vec<RingPoint> = Vec::with_capacity(5);
        for i in 0..5u32 {
            let token = u64::from(i) * span / 5;
            pts.push(RingPoint::new(token, i, "dc1", "r1"));
        }
        RingView::new(pts)
    }

    fn primary_idx(plan: &ReplicationPlan) -> u32 {
        match plan {
            ReplicationPlan::Successors { primary, .. } => primary.peer_idx,
            other @ ReplicationPlan::Topology(_) => {
                panic!("expected successors plan, got {other:?}")
            }
        }
    }

    fn successor_idxs(plan: &ReplicationPlan) -> Vec<u32> {
        match plan {
            ReplicationPlan::Successors { successors, .. } => {
                successors.iter().map(|t| t.peer_idx).collect()
            }
            other @ ReplicationPlan::Topology(_) => {
                panic!("expected successors plan, got {other:?}")
            }
        }
    }

    #[test]
    fn key_hashing_into_peer1_slot_returns_1_2_3() {
        let ring = five_peer_ring();
        // Hash 1 lands strictly above peer 0's token (0); the
        // owning slot is peer 1.
        let key_hash = 1;
        let plan = plan_replicas(
            &ring,
            key_hash,
            3,
            ReplicationStrategy::Successors,
            ConsistencyLevel::DcOne,
        );
        assert_eq!(primary_idx(&plan), 1);
        assert_eq!(successor_idxs(&plan), vec![2, 3]);
    }

    #[test]
    fn key_hashing_into_peer0_slot_returns_0_1_2() {
        let ring = five_peer_ring();
        // Token 0 itself is owned by peer 0 (binary search hits
        // exact match). This exercises the prompt's "hashes into
        // peer-0's slot; n_val=3 returns peers 0, 1, 2".
        let key_hash = 0;
        let plan = plan_replicas(
            &ring,
            key_hash,
            3,
            ReplicationStrategy::Successors,
            ConsistencyLevel::DcOne,
        );
        assert_eq!(primary_idx(&plan), 0);
        assert_eq!(successor_idxs(&plan), vec![1, 2]);
    }

    #[test]
    fn key_hashing_into_peer3_slot_wraps_around() {
        let ring = five_peer_ring();
        let span = u64::from(u32::MAX);
        // Choose a hash strictly above peer 2's token but at or
        // below peer 3's token. The owning slot is peer 3 and
        // the walk goes peer 4, then wraps to peer 0.
        let key_hash = 2 * span / 5 + 1;
        let plan = plan_replicas(
            &ring,
            key_hash,
            3,
            ReplicationStrategy::Successors,
            ConsistencyLevel::DcOne,
        );
        assert_eq!(primary_idx(&plan), 3);
        assert_eq!(successor_idxs(&plan), vec![4, 0]);
    }

    #[test]
    fn n_val_two_returns_two_peers() {
        let ring = five_peer_ring();
        let plan = plan_replicas(
            &ring,
            1,
            2,
            ReplicationStrategy::Successors,
            ConsistencyLevel::DcOne,
        );
        let flat = plan.into_replica_list();
        assert_eq!(flat.len(), 2);
        assert_eq!(flat[0].peer_idx, 1);
        assert_eq!(flat[1].peer_idx, 2);
    }

    #[test]
    fn n_val_larger_than_ring_caps_at_ring_size() {
        let ring = five_peer_ring();
        let plan = plan_replicas(
            &ring,
            1,
            10,
            ReplicationStrategy::Successors,
            ConsistencyLevel::DcOne,
        );
        let flat = plan.into_replica_list();
        assert_eq!(flat.len(), 5);
        let idxs: Vec<u32> = flat.iter().map(|t| t.peer_idx).collect();
        assert_eq!(idxs, vec![1, 2, 3, 4, 0]);
    }

    #[test]
    fn topology_strategy_returns_empty_passthrough() {
        let ring = five_peer_ring();
        let plan = plan_replicas(
            &ring,
            1,
            3,
            ReplicationStrategy::Topology,
            ConsistencyLevel::DcOne,
        );
        assert!(matches!(plan, ReplicationPlan::Topology(ref v) if v.is_empty()));
    }

    #[test]
    fn duplicate_peer_slots_are_deduplicated() {
        // Peer 0 has two ring points so a 3-replica walk skips
        // its second point in favour of the next distinct peer.
        let pts = vec![
            RingPoint::new(0, 0, "dc1", "r1"),
            RingPoint::new(100, 0, "dc1", "r1"),
            RingPoint::new(200, 1, "dc1", "r1"),
            RingPoint::new(300, 2, "dc1", "r1"),
        ];
        let ring = RingView::new(pts);
        let plan = plan_replicas(
            &ring,
            1,
            3,
            ReplicationStrategy::Successors,
            ConsistencyLevel::DcOne,
        );
        let flat = plan.into_replica_list();
        let idxs: Vec<u32> = flat.iter().map(|t| t.peer_idx).collect();
        assert_eq!(idxs, vec![0, 1, 2]);
    }

    #[test]
    fn empty_ring_yields_synthetic_primary() {
        let ring = RingView::new(Vec::new());
        let plan = plan_replicas(
            &ring,
            42,
            3,
            ReplicationStrategy::Successors,
            ConsistencyLevel::DcOne,
        );
        match plan {
            ReplicationPlan::Successors {
                primary,
                successors,
                ..
            } => {
                assert!(successors.is_empty());
                assert_eq!(primary.dc, "");
            }
            other @ ReplicationPlan::Topology(_) => {
                panic!("expected successors plan, got {other:?}")
            }
        }
    }

    #[test]
    fn from_wire_round_trips() {
        for s in [
            ReplicationStrategy::Topology,
            ReplicationStrategy::Successors,
        ] {
            assert_eq!(ReplicationStrategy::from_wire(s.to_wire()).unwrap(), s);
        }
        assert_eq!(
            ReplicationStrategy::from_wire(7),
            Err(ReplicationStrategyError::Unknown(7))
        );
    }

    #[test]
    fn primary_index_wraps_when_hash_exceeds_largest_token() {
        let ring = five_peer_ring();
        let huge = u64::from(u32::MAX) + 1_000_000;
        assert_eq!(ring.primary_index(huge), Some(0));
    }

    #[test]
    fn ring_view_counts_distinct_peers() {
        let pts = vec![
            RingPoint::new(0, 0, "dc1", "r1"),
            RingPoint::new(10, 0, "dc1", "r1"),
            RingPoint::new(20, 1, "dc1", "r1"),
        ];
        let r = RingView::new(pts);
        assert_eq!(r.peer_count(), 2);
    }
}