Fix replicator segment selection (#5046)

core/src/replicator.rs

@@ -28,7 +28,7 @@ use solana_sdk::client::{AsyncClient, SyncClient};
 use solana_sdk::hash::{Hash, Hasher};
 use solana_sdk::message::Message;
 use solana_sdk::signature::{Keypair, KeypairUtil, Signature};
-use solana_sdk::timing::{get_segment_from_slot, timestamp};
+use solana_sdk::timing::{get_complete_segment_from_slot, get_segment_from_slot, timestamp};
 use solana_sdk::transaction::Transaction;
 use solana_sdk::transport::TransportError;
 use solana_storage_api::storage_contract::StorageContract;
@@ -45,6 +45,8 @@ use std::sync::{Arc, RwLock};
 use std::thread::{sleep, spawn, JoinHandle};
 use std::time::Duration;
 
+static ENCRYPTED_FILENAME: &'static str = "ledger.enc";
+
 #[derive(Serialize, Deserialize)]
 pub enum ReplicatorRequest {
     GetSlotHeight(SocketAddr),
@@ -114,7 +116,8 @@ fn get_slot_from_signature(
         | (u64::from(signature_vec[1]) << 8)
         | (u64::from(signature_vec[1]) << 16)
         | (u64::from(signature_vec[2]) << 24);
-    let max_segment_index = get_segment_from_slot(storage_turn, slots_per_segment);
+    let max_segment_index =
+        get_complete_segment_from_slot(storage_turn, slots_per_segment).unwrap();
     segment_index %= max_segment_index as u64;
     segment_index * slots_per_segment
 }
@@ -514,7 +517,7 @@ impl Replicator {
 
     fn encrypt_ledger(meta: &mut ReplicatorMeta, blocktree: &Arc<Blocktree>) -> Result<()> {
         let ledger_path = Path::new(&meta.ledger_path);
-        meta.ledger_data_file_encrypted = ledger_path.join("ledger.enc");
+        meta.ledger_data_file_encrypted = ledger_path.join(ENCRYPTED_FILENAME);
 
         {
             let mut ivec = [0u8; 64];
@@ -709,7 +712,7 @@ impl Replicator {
                 previous_blockhash,
                 exit,
             )?;
-            if get_segment_from_slot(turn_slot, slots_per_segment) != 0 {
+            if get_complete_segment_from_slot(turn_slot, slots_per_segment).is_some() {
                 return Ok((blockhash, turn_slot));
             }
         }

sdk/src/timing.rs

@@ -67,8 +67,25 @@ pub fn timestamp() -> u64 {
     duration_as_ms(&now)
 }
 
-pub fn get_segment_from_slot(slot: Slot, slots_per_segment: u64) -> Segment {
+/// Converts a slot to a storage segment. Does not indicate that a segment is complete.
-    ((slot + (slots_per_segment - 1)) / slots_per_segment)
+pub fn get_segment_from_slot(rooted_slot: Slot, slots_per_segment: u64) -> Segment {
+    ((rooted_slot + (slots_per_segment - 1)) / slots_per_segment)
+}
+
+/// Given a slot returns the latest complete segment, if no segment could possibly be complete
+/// for a given slot it returns `None` (i.e if `slot < slots_per_segment`)
+pub fn get_complete_segment_from_slot(
+    rooted_slot: Slot,
+    slots_per_segment: u64,
+) -> Option<Segment> {
+    let current_segment = get_segment_from_slot(rooted_slot, slots_per_segment);
+    if current_segment == 1 {
+        None
+    } else if rooted_slot < (current_segment * slots_per_segment) {
+        Some(current_segment - 1)
+    } else {
+        Some(current_segment)
+    }
 }
 
 /// Slot is a unit of time given to a leader for encoding,
@@ -81,3 +98,29 @@ pub type Segment = u64;
 /// Epoch is a unit of time a given leader schedule is honored,
 ///  some number of Slots.  Use a u64 to count them.
 pub type Epoch = u64;
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn get_segments(slot: Slot, slots_per_segment: u64) -> (Segment, Segment) {
+        (
+            get_segment_from_slot(slot, slots_per_segment),
+            get_complete_segment_from_slot(slot, slots_per_segment).unwrap(),
+        )
+    }
+
+    #[test]
+    fn test_complete_segment_impossible() {
+        // slot < slots_per_segment so there can be no complete segments
+        assert_eq!(get_complete_segment_from_slot(5, 10), None);
+    }
+
+    #[test]
+    fn test_segment_conversion() {
+        let (current, complete) = get_segments(2048, 1024);
+        assert_eq!(current, complete);
+        let (current, complete) = get_segments(2049, 1024);
+        assert!(complete < current);
+    }
+}