Migrate raft_state from cbor to JSON (qdrant#2337)

* migrate raft_state to json

* review rename

* review rename

Author: generall

lib/storage/src/content_manager/consensus/persistent.rs

@@ -17,7 +17,10 @@ use crate::content_manager::consensus::entry_queue::{EntryApplyProgressQueue, En
 use crate::types::PeerAddressById;
 use crate::StorageError;
 
-const STATE_FILE_NAME: &str = "raft_state";
+// Deprecated, use `STATE_FILE_NAME` instead
+const STATE_FILE_NAME_CBOR: &str = "raft_state";
+
+const STATE_FILE_NAME: &str = "raft_state.json";
 
 /// State of the Raft consensus, which should be saved between restarts.
 /// State of the collections, aliases and transfers are stored as regular storage.
@@ -73,14 +76,23 @@ impl Persistent {
         first_peer: bool,
     ) -> Result<Self, StorageError> {
         create_dir_all(storage_path.as_ref())?;
-        let path = storage_path.as_ref().join(STATE_FILE_NAME);
-        let state = if path.exists() {
-            log::info!("Loading raft state from {}", path.display());
-            Self::load(path)?
+        let path_legacy = storage_path.as_ref().join(STATE_FILE_NAME_CBOR);
+        let path_json = storage_path.as_ref().join(STATE_FILE_NAME);
+        let state = if path_json.exists() {
+            log::info!("Loading raft state from {}", path_json.display());
+            Self::load_json(path_json)?
+        } else if path_legacy.exists() {
+            log::info!("Loading raft state from {}", path_legacy.display());
+            let mut state = Self::load(path_legacy)?;
+            // migrate to json
+            state.path = path_json;
+            state.save()?;
+            state
         } else {
-            log::info!("Initializing new raft state at {}", path.display());
-            Self::init(path, first_peer)?
+            log::info!("Initializing new raft state at {}", path_json.display());
+            Self::init(path_json, first_peer)?
         };
+
         log::debug!("State: {:?}", state);
         Ok(state)
     }
@@ -192,10 +204,17 @@ impl Persistent {
         Ok(state)
     }
 
+    fn load_json(path: PathBuf) -> Result<Self, StorageError> {
+        let file = File::open(&path)?;
+        let mut state: Self = serde_json::from_reader(&file)?;
+        state.path = path;
+        Ok(state)
+    }
+
     pub fn save(&self) -> Result<(), StorageError> {
         let result = AtomicFile::new(&self.path, AllowOverwrite).write(|file| {
             let writer = BufWriter::new(file);
-            serde_cbor::to_writer(writer, self)
+            serde_json::to_writer(writer, self)
         });
         log::trace!("Saved state: {:?}", self);
         self.dirty.store(result.is_err(), Ordering::Relaxed);

lib/storage/src/content_manager/errors.rs

@@ -193,6 +193,15 @@ impl From<serde_cbor::Error> for StorageError {
     }
 }
 
+impl From<serde_json::Error> for StorageError {
+    fn from(err: serde_json::Error) -> Self {
+        StorageError::ServiceError {
+            description: format!("json (de)serialization error: {err}"),
+            backtrace: Some(Backtrace::force_capture().to_string()),
+        }
+    }
+}
+
 impl From<prost::EncodeError> for StorageError {
     fn from(err: prost::EncodeError) -> Self {
         StorageError::ServiceError {