*: fix node bootstrap not idempotent

src/raftstore/store/bootstrap.rs

@@ -55,24 +55,25 @@ pub fn bootstrap_store(engine: &DB, cluster_id: u64, store_id: u64) -> Result<()
     engine.put_msg(&ident_key, &ident)
 }
 
-// Write first region meta.
-pub fn write_region(engine: &DB, region: &metapb::Region) -> Result<()> {
+// Write first region meta and prepare state.
+pub fn write_prepare_bootstrap(engine: &DB, region: &metapb::Region) -> Result<()> {
     let mut state = RegionLocalState::new();
     state.set_region(region.clone());
 
     let wb = WriteBatch::new();
     try!(wb.put_msg(&keys::region_state_key(region.get_id()), &state));
     try!(write_initial_state(engine, &wb, region.get_id()));
+    try!(wb.put_msg(&keys::prepare_bootstrap_key(), region));
     try!(engine.write(wb));
     Ok(())
 }
 
-// Clear first region meta.
-pub fn clear_region(engine: &DB, region_id: u64) -> Result<()> {
+// Clear first region meta and prepare state.
+pub fn clear_prepare_bootstrap(engine: &DB, region_id: u64) -> Result<()> {
     let wb = WriteBatch::new();
 
     try!(wb.delete(&keys::region_state_key(region_id)));
-
+    try!(wb.delete(&keys::prepare_bootstrap_key()));
     // should clear raft initial state too.
     let raft_cf = try!(rocksdb::get_cf_handle(engine, CF_RAFT));
     try!(wb.delete_cf(raft_cf, &keys::raft_state_key(region_id)));
@@ -82,12 +83,18 @@ pub fn clear_region(engine: &DB, region_id: u64) -> Result<()> {
     Ok(())
 }
 
-// Bootstrap first region.
-pub fn bootstrap_region(engine: &DB,
-                        store_id: u64,
-                        region_id: u64,
-                        peer_id: u64)
-                        -> Result<metapb::Region> {
+// Clear prepare state
+pub fn clear_prepare_bootstrap_state(engine: &DB) -> Result<()> {
+    try!(engine.delete(&keys::prepare_bootstrap_key()));
+    Ok(())
+}
+
+// Prepare bootstrap.
+pub fn prepare_bootstrap(engine: &DB,
+                         store_id: u64,
+                         region_id: u64,
+                         peer_id: u64)
+                         -> Result<metapb::Region> {
     let mut region = metapb::Region::new();
     region.set_id(region_id);
     region.set_start_key(keys::EMPTY_KEY.to_vec());
@@ -100,7 +107,7 @@ pub fn bootstrap_region(engine: &DB,
     peer.set_id(peer_id);
     region.mut_peers().push(peer);
 
-    try!(write_region(engine, &region));
+    try!(write_prepare_bootstrap(engine, &region));
 
     Ok(region)
 }
@@ -124,12 +131,14 @@ mod tests {
         assert!(bootstrap_store(&engine, 1, 1).is_ok());
         assert!(bootstrap_store(&engine, 1, 1).is_err());
 
-        assert!(bootstrap_region(&engine, 1, 1, 1).is_ok());
+        assert!(prepare_bootstrap(&engine, 1, 1, 1).is_ok());
         assert!(engine.get_value(&keys::region_state_key(1)).unwrap().is_some());
+        assert!(engine.get_value(&keys::prepare_bootstrap_key()).unwrap().is_some());
         assert!(engine.get_value_cf(CF_RAFT, &keys::raft_state_key(1)).unwrap().is_some());
         assert!(engine.get_value_cf(CF_RAFT, &keys::apply_state_key(1)).unwrap().is_some());
 
-        assert!(clear_region(&engine, 1).is_ok());
+        assert!(clear_prepare_bootstrap_state(&engine).is_ok());
+        assert!(clear_prepare_bootstrap(&engine, 1).is_ok());
         assert!(is_range_empty(&engine,
                                CF_DEFAULT,
                                &keys::region_meta_prefix(1),

src/raftstore/store/keys.rs

@@ -37,6 +37,7 @@ pub const DATA_MAX_KEY: &'static [u8] = &[DATA_PREFIX + 1];
 
 // Following keys are all local keys, so the first byte must be 0x01.
 pub const STORE_IDENT_KEY: &'static [u8] = &[LOCAL_PREFIX, 0x01];
+pub const PREPARE_BOOTSTRAP_KEY: &'static [u8] = &[LOCAL_PREFIX, 0x02];
 // We save two types region data in DB, for raft and other meta data.
 // When the store starts, we should iterate all region meta data to
 // construct peer, no need to travel large raft data, so we separate them
@@ -61,6 +62,10 @@ pub fn store_ident_key() -> Vec<u8> {
     STORE_IDENT_KEY.to_vec()
 }
 
+pub fn prepare_bootstrap_key() -> Vec<u8> {
+    PREPARE_BOOTSTRAP_KEY.to_vec()
+}
+
 fn make_region_id_key(region_id: u64, suffix: u8, extra_cap: usize) -> Vec<u8> {
     let mut key = Vec::with_capacity(REGION_RAFT_PREFIX_KEY.len() + mem::size_of::<u64>() +
                                      mem::size_of::<u8>() +

src/raftstore/store/mod.rs

@@ -34,7 +34,8 @@ pub use self::store::{StoreChannel, Store, create_event_loop};
 pub use self::config::Config;
 pub use self::transport::Transport;
 pub use self::peer::Peer;
-pub use self::bootstrap::{bootstrap_store, bootstrap_region, write_region, clear_region};
+pub use self::bootstrap::{bootstrap_store, prepare_bootstrap, write_prepare_bootstrap,
+                          clear_prepare_bootstrap, clear_prepare_bootstrap_state};
 pub use self::engine::{Peekable, Iterable, Mutable};
 pub use self::peer_storage::{PeerStorage, do_snapshot, SnapState, RAFT_INIT_LOG_TERM,
                              RAFT_INIT_LOG_INDEX};

src/raftstore/store/peer_storage.rs

@@ -990,7 +990,7 @@ mod test {
         let db = new_engine(path.path().to_str().unwrap(), ALL_CFS).unwrap();
         let db = Arc::new(db);
         bootstrap::bootstrap_store(&db, 1, 1).expect("");
-        let region = bootstrap::bootstrap_region(&db, 1, 1, 1).expect("");
+        let region = bootstrap::prepare_bootstrap(&db, 1, 1, 1).expect("");
         PeerStorage::new(db, &region, sched, "".to_owned()).unwrap()
     }
 

src/server/node.rs

@@ -116,11 +116,11 @@ impl<C> Node<C>
         }
 
         self.store.set_id(store_id);
-
+        try!(self.check_prepare_bootstrap_cluster(&engine));
         if !bootstrapped {
             // cluster is not bootstrapped, and we choose first store to bootstrap
-            // first region.
-            let region = try!(self.bootstrap_first_region(&engine, store_id));
+            // prepare bootstrap.
+            let region = try!(self.prepare_bootstrap_cluster(&engine, store_id));
             try!(self.bootstrap_cluster(&engine, region));
         }
 
@@ -181,7 +181,7 @@ impl<C> Node<C>
         Ok(store_id)
     }
 
-    fn bootstrap_first_region(&self, engine: &DB, store_id: u64) -> Result<metapb::Region> {
+    fn prepare_bootstrap_cluster(&self, engine: &DB, store_id: u64) -> Result<metapb::Region> {
         let region_id = try!(self.alloc_id());
         info!("alloc first region id {} for cluster {}, store {}",
               region_id,
@@ -192,21 +192,66 @@ impl<C> Node<C>
               peer_id,
               region_id);
 
-        let region = try!(store::bootstrap_region(engine, store_id, region_id, peer_id));
+        let region = try!(store::prepare_bootstrap(engine, store_id, region_id, peer_id));
         Ok(region)
     }
 
+    fn check_region_epoch(&self, region: &metapb::Region, other: &metapb::Region) -> Result<()> {
+        let epoch = region.get_region_epoch();
+        let other_epoch = other.get_region_epoch();
+        if epoch.get_conf_ver() != other_epoch.get_conf_ver() {
+            return Err(box_err!("region conf_ver inconsist: {} with {}",
+                                epoch.get_conf_ver(),
+                                other_epoch.get_conf_ver()));
+        }
+        if epoch.get_version() != other_epoch.get_version() {
+            return Err(box_err!("region version inconsist: {} with {}",
+                                epoch.get_version(),
+                                other_epoch.get_version()));
+        }
+        Ok(())
+    }
+
+    fn check_prepare_bootstrap_cluster(&self, engine: &DB) -> Result<()> {
+        let res = try!(engine.get_msg::<metapb::Region>(&keys::prepare_bootstrap_key()));
+        if res.is_none() {
+            return Ok(());
+        }
+
+        let first_region = res.unwrap();
+        for _ in 0..MAX_CHECK_CLUSTER_BOOTSTRAPPED_RETRY_COUNT {
+            match self.pd_client.get_region(b"") {
+                Ok(region) => {
+                    if region.get_id() == first_region.get_id() {
+                        try!(self.check_region_epoch(&region, &first_region));
+                        try!(store::clear_prepare_bootstrap_state(engine));
+                    } else {
+                        try!(store::clear_prepare_bootstrap(engine, region.get_id()));
+                    }
+                    return Ok(());
+                }
+
+                Err(e) => {
+                    warn!("check cluster prepare bootstrapped failed: {:?}", e);
+                }
+            }
+            thread::sleep(Duration::from_secs(CHECK_CLUSTER_BOOTSTRAPPED_RETRY_SECONDS));
+        }
+        Err(box_err!("check cluster prepare bootstrapped failed"))
+    }
+
     fn bootstrap_cluster(&mut self, engine: &DB, region: metapb::Region) -> Result<()> {
         let region_id = region.get_id();
         match self.pd_client.bootstrap_cluster(self.store.clone(), region) {
             Err(PdError::ClusterBootstrapped(_)) => {
                 error!("cluster {} is already bootstrapped", self.cluster_id);
-                try!(store::clear_region(engine, region_id));
+                try!(store::clear_prepare_bootstrap(engine, region_id));
                 Ok(())
             }
             // TODO: should we clean region for other errors too?
             Err(e) => panic!("bootstrap cluster {} err: {:?}", self.cluster_id, e),
             Ok(_) => {
+                try!(store::clear_prepare_bootstrap_state(engine));
                 info!("bootstrap cluster {} ok", self.cluster_id);
                 Ok(())
             }

tests/raftstore/cluster.rs

@@ -307,6 +307,28 @@ impl<T: Simulator> Cluster<T> {
         self.leaders.get(&region_id).cloned()
     }
 
+    pub fn check_regions_number(&self, len: u32) {
+        assert_eq!(self.pd_client.get_regions_number().unwrap(), len)
+    }
+
+    // For test when a node is already bootstraped the cluster with the first region
+    // But another node may request bootstrap at same time and get is_bootstrap false
+    // Add Region but not set bootstrap to true
+    pub fn add_first_region(&self) -> Result<()> {
+        let mut region = metapb::Region::new();
+        let region_id = self.pd_client.alloc_id().unwrap();
+        let peer_id = self.pd_client.alloc_id().unwrap();
+        region.set_id(region_id);
+        region.set_start_key(keys::EMPTY_KEY.to_vec());
+        region.set_end_key(keys::EMPTY_KEY.to_vec());
+        region.mut_region_epoch().set_version(1);
+        region.mut_region_epoch().set_conf_ver(1);
+        let peer = new_peer(peer_id, peer_id);
+        region.mut_peers().push(peer.clone());
+        self.pd_client.add_region(&region);
+        Ok(())
+    }
+
     // Multiple nodes with fixed node id, like node 1, 2, .. 5,
     // First region 1 is in all stores with peer 1, 2, .. 5.
     // Peer 1 is in node 1, store 1, etc.
@@ -330,7 +352,7 @@ impl<T: Simulator> Cluster<T> {
         }
 
         for engine in self.engines.values() {
-            try!(write_region(&engine, &region));
+            try!(write_prepare_bootstrap(&engine, &region));
         }
 
         self.bootstrap_cluster(region);
@@ -350,7 +372,7 @@ impl<T: Simulator> Cluster<T> {
         }
 
         let node_id = 1;
-        let region = bootstrap_region(&self.engines[&node_id], 1, 1, 1).unwrap();
+        let region = prepare_bootstrap(&self.engines[&node_id], 1, 1, 1).unwrap();
         let rid = region.get_id();
         self.bootstrap_cluster(region);
         rid

tests/raftstore/mod.rs

@@ -34,3 +34,4 @@ mod test_snap;
 mod test_region_heartbeat;
 mod test_stale_peer;
 mod test_lease_read;
+mod test_bootstrap;

tests/raftstore/pd.rs

@@ -57,6 +57,7 @@ struct Cluster {
 
     down_peers: HashMap<u64, pdpb::PeerStats>,
     pending_peers: HashMap<u64, metapb::Peer>,
+    is_bootstraped: bool,
 }
 
 impl Cluster {
@@ -76,6 +77,7 @@ impl Cluster {
             split_count: 0,
             down_peers: HashMap::new(),
             pending_peers: HashMap::new(),
+            is_bootstraped: false,
         }
     }
 
@@ -96,6 +98,11 @@ impl Cluster {
         self.stores.insert(store_id, s);
 
         self.add_region(&region);
+        self.is_bootstraped = true;
+    }
+
+    fn set_bootstrap(&mut self, is_bootstraped: bool) {
+        self.is_bootstraped = is_bootstraped
     }
 
     // We don't care cluster id here, so any value like 0 in tests is ok.
@@ -131,6 +138,10 @@ impl Cluster {
         self.stores.values().map(|s| s.store.clone()).collect()
     }
 
+    fn get_regions_number(&self) -> Result<u32> {
+        Ok(self.regions.len() as u32)
+    }
+
     fn add_region(&mut self, region: &metapb::Region) {
         let end_key = enc_end_key(region);
         assert!(self.regions.insert(end_key.clone(), region.clone()).is_none());
@@ -362,6 +373,10 @@ impl TestPdClient {
         Ok(())
     }
 
+    fn is_regions_empty(&self) -> Result<(bool)> {
+        Ok(self.cluster.rl().regions.is_empty())
+    }
+
     // Set a customized rule to overwrite default max peer count check rule.
     pub fn set_rule(&self, rule: Rule) {
         self.cluster.wl().rule = Some(rule);
@@ -375,7 +390,9 @@ impl TestPdClient {
     pub fn get_region_epoch(&self, region_id: u64) -> metapb::RegionEpoch {
         self.get_region_by_id(region_id).wait().unwrap().unwrap().take_region_epoch()
     }
-
+    pub fn get_regions_number(&self) -> Result<(u32)> {
+        self.cluster.rl().get_regions_number()
+    }
     // Set an empty rule which nothing to do to disable default max peer count
     // check rule, we can use reset_rule to enable default again.
     pub fn disable_default_rule(&self) {
@@ -423,6 +440,9 @@ impl TestPdClient {
             .unwrap();
         panic!("region {:?} has peer {:?}", region, peer);
     }
+    pub fn add_region(&self, region: &metapb::Region) {
+        self.cluster.wl().add_region(region)
+    }
 
     pub fn add_peer(&self, region_id: u64, peer: metapb::Peer) {
         self.set_rule(box move |region: &metapb::Region, _: &metapb::Peer| {
@@ -503,7 +523,8 @@ impl PdClient for TestPdClient {
     }
 
     fn bootstrap_cluster(&self, store: metapb::Store, region: metapb::Region) -> Result<()> {
-        if self.is_cluster_bootstrapped().unwrap() {
+        if self.is_cluster_bootstrapped().unwrap() || !self.is_regions_empty().unwrap() {
+            self.cluster.wl().set_bootstrap(true);
             return Err(Error::ClusterBootstrapped(self.cluster_id));
         }
 
@@ -513,7 +534,7 @@ impl PdClient for TestPdClient {
     }
 
     fn is_cluster_bootstrapped(&self) -> Result<bool> {
-        Ok(!self.cluster.rl().stores.is_empty())
+        Ok(self.cluster.rl().is_bootstraped)
     }
 
     fn alloc_id(&self) -> Result<u64> {

tests/raftstore/test_bootstrap.rs

@@ -0,0 +1,35 @@
+// Copyright 2017 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use super::cluster::{Cluster, Simulator};
+use super::node::new_node_cluster;
+use super::util::*;
+
+fn test_bootstrap_idempotent<T: Simulator>(cluster: &mut Cluster<T>) {
+    // assume that there is a node  bootstrap the cluster and add region in pd successfully
+    cluster.add_first_region().unwrap();
+    // now  at same time start the another node, and will recive cluster is not bootstrap
+    // try to bootstrap with a new region
+    cluster.start();
+    cluster.check_regions_number(1);
+    cluster.shutdown();
+    sleep_ms(500);
+    cluster.start();
+    cluster.check_regions_number(1);
+}
+
+#[test]
+fn test_node_bootstrap_idempotent() {
+    let mut cluster = new_node_cluster(0, 3);
+    test_bootstrap_idempotent(&mut cluster);
+}