Wait for operation commit (#436)

* Wait for operation commit * Review: Option with timeout, return execution err to API * Review: Add comment about error forwarding in ???
2026-08-02 16:10:58 -05:00 · 2022-04-06 18:00:44 +03:00
parent 772e8b67bd
commit aeea31dfa4
10 changed files with 184 additions and 78 deletions
--- a/.github/workflows/integration-tests.yml
+++ b/.github/workflows/integration-tests.yml
@@ -30,3 +30,24 @@ jobs:
        run: |
          ./tests/integration-tests.sh
        shell: bash
+
+  test-consensus:
+
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Install minimal stable
+        uses: actions-rs/toolchain@v1
+        with:
+          profile: minimal
+          toolchain: stable
+      - uses: actions/checkout@v3
+      - uses: Swatinem/rust-cache@v1
+      - name: Install dependencies
+        run: sudo apt-get install clang
+      - name: Build
+        run: cargo build --features consensus
+      - name: Run integration tests
+        run: |
+          ./tests/integration-tests.sh
+        shell: bash
--- a/lib/collection/src/operations/config_diff.rs
+++ b/lib/collection/src/operations/config_diff.rs
@@ -22,7 +22,7 @@ pub trait DiffConfig<T: DeserializeOwned + Serialize> {
    }
 }

-#[derive(Debug, Deserialize, Serialize, JsonSchema, Copy, Clone, PartialEq, Eq, Merge)]
+#[derive(Debug, Deserialize, Serialize, JsonSchema, Copy, Clone, PartialEq, Eq, Merge, Hash)]
 #[serde(rename_all = "snake_case")]
 pub struct HnswConfigDiff {
    /// Number of edges per node in the index graph. Larger the value - more accurate the search, more space required.
@@ -35,7 +35,7 @@ pub struct HnswConfigDiff {
    pub full_scan_threshold: Option<usize>,
 }

-#[derive(Debug, Deserialize, Serialize, JsonSchema, Clone, Merge)]
+#[derive(Debug, Deserialize, Serialize, JsonSchema, Clone, Merge, PartialEq, Eq, Hash)]
 pub struct WalConfigDiff {
    /// Size of a single WAL segment in MB
    pub wal_capacity_mb: Option<usize>,
@@ -78,6 +78,37 @@ pub struct OptimizersConfigDiff {
    pub max_optimization_threads: Option<usize>,
 }

+impl std::hash::Hash for OptimizersConfigDiff {
+    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+        self.deleted_threshold.map(f64::to_le_bytes).hash(state);
+        self.vacuum_min_vector_number.hash(state);
+        self.default_segment_number.hash(state);
+        self.max_segment_size.hash(state);
+        self.memmap_threshold.hash(state);
+        self.indexing_threshold.hash(state);
+        self.payload_indexing_threshold.hash(state);
+        self.flush_interval_sec.hash(state);
+        self.max_optimization_threads.hash(state);
+    }
+}
+
+impl PartialEq for OptimizersConfigDiff {
+    fn eq(&self, other: &Self) -> bool {
+        self.deleted_threshold.map(f64::to_le_bytes)
+            == other.deleted_threshold.map(f64::to_le_bytes)
+            && self.vacuum_min_vector_number == other.vacuum_min_vector_number
+            && self.default_segment_number == other.default_segment_number
+            && self.max_segment_size == other.max_segment_size
+            && self.memmap_threshold == other.memmap_threshold
+            && self.indexing_threshold == other.indexing_threshold
+            && self.payload_indexing_threshold == other.payload_indexing_threshold
+            && self.flush_interval_sec == other.flush_interval_sec
+            && self.max_optimization_threads == other.max_optimization_threads
+    }
+}
+
+impl Eq for OptimizersConfigDiff {}
+
 impl DiffConfig<HnswConfig> for HnswConfigDiff {}

 impl DiffConfig<OptimizersConfig> for OptimizersConfigDiff {}
--- a/lib/segment/src/types.rs
+++ b/lib/segment/src/types.rs
@@ -72,7 +72,9 @@ impl FromStr for ExtendedPointId {
 pub type PointIdType = ExtendedPointId;

 /// Type of internal tags, build from payload
-#[derive(Debug, Deserialize, Serialize, JsonSchema, Clone, Copy, FromPrimitive, PartialEq, Eq)]
+#[derive(
+    Debug, Deserialize, Serialize, JsonSchema, Clone, Copy, FromPrimitive, PartialEq, Eq, Hash,
+)]
 /// Distance function types used to compare vectors
 pub enum Distance {
    /// https://en.wikipedia.org/wiki/Cosine_similarity
--- a/lib/storage/src/content_manager/collection_meta_ops.rs
+++ b/lib/storage/src/content_manager/collection_meta_ops.rs
@@ -9,35 +9,35 @@ use serde::{Deserialize, Serialize};

 /// Create alternative name for a collection.
 /// Collection will be available under both names for search, retrieve,
-#[derive(Debug, Deserialize, Serialize, JsonSchema)]
+#[derive(Debug, Deserialize, Serialize, JsonSchema, PartialEq, Eq, Hash)]
 #[serde(rename_all = "snake_case")]
 pub struct CreateAlias {
    pub collection_name: String,
    pub alias_name: String,
 }

-#[derive(Debug, Deserialize, Serialize, JsonSchema)]
+#[derive(Debug, Deserialize, Serialize, JsonSchema, PartialEq, Eq, Hash)]
 #[serde(rename_all = "snake_case")]
 pub struct CreateAliasOperation {
    pub create_alias: CreateAlias,
 }

 /// Delete alias if exists
-#[derive(Debug, Deserialize, Serialize, JsonSchema)]
+#[derive(Debug, Deserialize, Serialize, JsonSchema, PartialEq, Eq, Hash)]
 #[serde(rename_all = "snake_case")]
 pub struct DeleteAlias {
    pub alias_name: String,
 }

 /// Delete alias if exists
-#[derive(Debug, Deserialize, Serialize, JsonSchema)]
+#[derive(Debug, Deserialize, Serialize, JsonSchema, PartialEq, Eq, Hash)]
 #[serde(rename_all = "snake_case")]
 pub struct DeleteAliasOperation {
    pub delete_alias: DeleteAlias,
 }

 /// Change alias to a new one
-#[derive(Debug, Deserialize, Serialize, JsonSchema)]
+#[derive(Debug, Deserialize, Serialize, JsonSchema, PartialEq, Eq, Hash)]
 #[serde(rename_all = "snake_case")]
 pub struct RenameAlias {
    pub old_alias_name: String,
@@ -45,14 +45,14 @@ pub struct RenameAlias {
 }

 /// Change alias to a new one
-#[derive(Debug, Deserialize, Serialize, JsonSchema)]
+#[derive(Debug, Deserialize, Serialize, JsonSchema, PartialEq, Eq, Hash)]
 #[serde(rename_all = "snake_case")]
 pub struct RenameAliasOperation {
    pub rename_alias: RenameAlias,
 }

 /// Group of all the possible operations related to collection aliases
-#[derive(Debug, Deserialize, Serialize, JsonSchema)]
+#[derive(Debug, Deserialize, Serialize, JsonSchema, PartialEq, Eq, Hash)]
 #[serde(rename_all = "snake_case")]
 #[serde(untagged)]
 pub enum AliasOperations {
@@ -80,7 +80,7 @@ impl From<RenameAlias> for AliasOperations {
 }

 /// Operation for creating new collection and (optionally) specify index params
-#[derive(Debug, Deserialize, Serialize, JsonSchema)]
+#[derive(Debug, Deserialize, Serialize, JsonSchema, PartialEq, Eq, Hash)]
 #[serde(rename_all = "snake_case")]
 pub struct CreateCollection {
    pub vector_size: usize,
@@ -101,7 +101,7 @@ pub const fn default_shard_number() -> u32 {
 }

 /// Operation for creating new collection and (optionally) specify index params
-#[derive(Debug, Deserialize, Serialize, JsonSchema)]
+#[derive(Debug, Deserialize, Serialize, JsonSchema, PartialEq, Eq, Hash)]
 #[serde(rename_all = "snake_case")]
 pub struct CreateCollectionOperation {
    pub collection_name: String,
@@ -110,7 +110,7 @@ pub struct CreateCollectionOperation {
 }

 /// Operation for updating parameters of the existing collection
-#[derive(Debug, Deserialize, Serialize, JsonSchema)]
+#[derive(Debug, Deserialize, Serialize, JsonSchema, PartialEq, Eq, Hash)]
 #[serde(rename_all = "snake_case")]
 pub struct UpdateCollection {
    /// Custom params for Optimizers.  If none - values from service configuration file are used.
@@ -119,7 +119,7 @@ pub struct UpdateCollection {
 }

 /// Operation for updating parameters of the existing collection
-#[derive(Debug, Deserialize, Serialize, JsonSchema)]
+#[derive(Debug, Deserialize, Serialize, JsonSchema, PartialEq, Eq, Hash)]
 #[serde(rename_all = "snake_case")]
 pub struct UpdateCollectionOperation {
    pub collection_name: String,
@@ -130,19 +130,19 @@ pub struct UpdateCollectionOperation {
 /// Operation for performing changes of collection aliases.
 /// Alias changes are atomic, meaning that no collection modifications can happen between
 /// alias operations.
-#[derive(Debug, Deserialize, Serialize, JsonSchema)]
+#[derive(Debug, Deserialize, Serialize, JsonSchema, PartialEq, Eq, Hash)]
 #[serde(rename_all = "snake_case")]
 pub struct ChangeAliasesOperation {
    pub actions: Vec<AliasOperations>,
 }

 /// Operation for deleting collection with given name
-#[derive(Debug, Deserialize, Serialize, JsonSchema)]
+#[derive(Debug, Deserialize, Serialize, JsonSchema, PartialEq, Eq, Hash)]
 #[serde(rename_all = "snake_case")]
 pub struct DeleteCollectionOperation(pub String);

 /// Enumeration of all possible collection update operations
-#[derive(Debug, Deserialize, Serialize, JsonSchema)]
+#[derive(Debug, Deserialize, Serialize, JsonSchema, PartialEq, Eq, Hash)]
 #[serde(rename_all = "snake_case")]
 pub enum CollectionMetaOperations {
    CreateCollection(CreateCollectionOperation),
--- a/lib/storage/src/content_manager/errors.rs
+++ b/lib/storage/src/content_manager/errors.rs
@@ -81,6 +81,14 @@ impl<T> From<std::sync::mpsc::SendError<T>> for StorageError {
    }
 }

+impl From<tokio::sync::oneshot::error::RecvError> for StorageError {
+    fn from(err: tokio::sync::oneshot::error::RecvError) -> Self {
+        StorageError::ServiceError {
+            description: format!("Channel sender dropped: {}", err),
+        }
+    }
+}
+
 #[cfg(feature = "consensus")]
 impl From<serde_cbor::Error> for StorageError {
    fn from(err: serde_cbor::Error) -> Self {
--- a/lib/storage/src/content_manager/toc.rs
+++ b/lib/storage/src/content_manager/toc.rs
@@ -1,8 +1,10 @@
 use std::collections::HashMap;
 use std::fs::{create_dir_all, read_dir, remove_dir_all};
 use std::num::NonZeroU32;
+use std::ops::Deref;
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
+use std::time::Duration;

 use tokio::runtime::Runtime;
 use tokio::sync::{RwLock, RwLockReadGuard};
@@ -30,14 +32,15 @@ use crate::content_manager::{
 use crate::types::StorageConfig;
 use collection::collection_manager::collection_managers::CollectionSearcher;
 use collection::collection_manager::simple_collection_searcher::SimpleCollectionSearcher;
-
 use collection::shard::ShardId;
+
 #[cfg(feature = "consensus")]
 use raft::{
    eraftpb::{Entry as RaftEntry, Snapshot as RaftSnapshot},
    RaftState,
 };
-use std::ops::Deref;
+#[cfg(feature = "consensus")]
+use tokio::sync::oneshot;
 #[cfg(feature = "consensus")]
 use wal::Wal;

@@ -47,6 +50,8 @@ pub use consensus::TableOfContentRef;
 const COLLECTIONS_DIR: &str = "collections";
 #[cfg(feature = "consensus")]
 const COLLECTIONS_META_WAL_DIR: &str = "collections_meta_wal";
+#[cfg(feature = "consensus")]
+const DEFAULT_META_OP_WAIT: Duration = Duration::from_secs(10);

 /// The main object of the service. It holds all objects, required for proper functioning.
 /// In most cases only one `TableOfContent` is enough for service. It is created only once during
@@ -65,6 +70,10 @@ pub struct TableOfContent {
    raft_state: Arc<std::sync::Mutex<RaftState>>,
    #[cfg(feature = "consensus")]
    propose_sender: Option<std::sync::Mutex<std::sync::mpsc::Sender<Vec<u8>>>>,
+    #[cfg(feature = "consensus")]
+    on_meta_op_apply: std::sync::Mutex<
+        HashMap<CollectionMetaOperations, oneshot::Sender<Result<bool, StorageError>>>,
+    >,
 }

 impl TableOfContent {
@@ -125,6 +134,8 @@ impl TableOfContent {
            raft_state: Arc::new(std::sync::Mutex::new(RaftState::default())),
            #[cfg(feature = "consensus")]
            propose_sender: None,
+            #[cfg(feature = "consensus")]
+            on_meta_op_apply: std::sync::Mutex::new(HashMap::new()),
        }
    }

@@ -321,13 +332,18 @@ impl TableOfContent {
        Ok(true)
    }

+    /// If `wait_timeout` is not supplied - then default duration will be used.
+    /// This function needs to be called from a runtime with timers enabled.
+    #[allow(unused_variables)]
    pub async fn submit_collection_operation(
        &self,
        operation: CollectionMetaOperations,
+        wait_timeout: Option<Duration>,
    ) -> Result<bool, StorageError> {
        #[cfg(feature = "consensus")]
        {
-            self.propose_collection_meta_op(operation)
+            self.propose_collection_meta_op(operation, wait_timeout.unwrap_or(DEFAULT_META_OP_WAIT))
+                .await
        }
        #[cfg(not(feature = "consensus"))]
        {
@@ -336,17 +352,35 @@ impl TableOfContent {
    }

    #[cfg(feature = "consensus")]
-    fn propose_collection_meta_op(
+    async fn propose_collection_meta_op(
        &self,
        operation: CollectionMetaOperations,
+        wait_timeout: Duration,
    ) -> Result<bool, StorageError> {
-        match &self.propose_sender {
-            Some(sender) => sender.lock()?.send(serde_cbor::to_vec(&operation)?)?,
-            None => log::error!(
-                "Cannot submit collection meta operation proposal: no sender supplied to ToC"
-            ),
-        }
-        Ok(true)
+        let propose_sender = match &self.propose_sender {
+            Some(sender) => sender,
+            None => {
+                log::error!(
+                    "Cannot submit collection meta operation proposal: no sender supplied to ToC"
+                );
+                return Ok(true);
+            }
+        };
+        let serialized = serde_cbor::to_vec(&operation)?;
+        let (sender, receiver) = oneshot::channel();
+        self.on_meta_op_apply.lock()?.insert(operation, sender);
+        propose_sender.lock()?.send(serialized)?;
+        Ok(tokio::time::timeout(wait_timeout, receiver)
+            .await
+            .map_err(
+                |_: tokio::time::error::Elapsed| StorageError::ServiceError {
+                    description: format!(
+                        "Waiting for collection meta operation commit failed. Timeout set at: {} seconds",
+                        wait_timeout.as_secs_f64()
+                    ),
+                },
+                // ??? - forwards 3 possible errors: timeout, sender dropped, operation failed
+            )???)
    }

    async fn perform_collection_meta_op(
@@ -561,8 +595,16 @@ impl TableOfContent {
    #[cfg(feature = "consensus")]
    pub fn apply_entry(&self, entry: &RaftEntry) -> Result<bool, StorageError> {
        let operation: CollectionMetaOperations = entry.try_into()?;
-        self.collection_management_runtime
-            .block_on(self.perform_collection_meta_op(operation))
+        let on_apply = self.on_meta_op_apply.lock()?.remove(&operation);
+        let result = self
+            .collection_management_runtime
+            .block_on(self.perform_collection_meta_op(operation));
+        if let Some(on_apply) = on_apply {
+            if on_apply.send(result.clone()).is_err() {
+                log::warn!("Failed to notify on collection meta operation completion.")
+            }
+        }
+        result
    }

    #[cfg(feature = "consensus")]
--- a/lib/storage/tests/alias_tests.rs
+++ b/lib/storage/tests/alias_tests.rs
@@ -43,42 +43,39 @@ mod tests {
        let toc = TableOfContent::new(&config, runtime);

        handle
-            .block_on(
-                toc.submit_collection_operation(CollectionMetaOperations::CreateCollection(
-                    CreateCollectionOperation {
-                        collection_name: "test".to_string(),
-                        create_collection: CreateCollection {
-                            vector_size: 10,
-                            distance: Distance::Cosine,
-                            hnsw_config: None,
-                            wal_config: None,
-                            optimizers_config: None,
-                            shard_number: 1,
-                        },
+            .block_on(toc.submit_collection_operation(
+                CollectionMetaOperations::CreateCollection(CreateCollectionOperation {
+                    collection_name: "test".to_string(),
+                    create_collection: CreateCollection {
+                        vector_size: 10,
+                        distance: Distance::Cosine,
+                        hnsw_config: None,
+                        wal_config: None,
+                        optimizers_config: None,
+                        shard_number: 1,
                    },
-                )),
-            )
+                }),
+                None,
+            ))
            .unwrap();

        handle
-            .block_on(
-                toc.submit_collection_operation(CollectionMetaOperations::ChangeAliases(
-                    ChangeAliasesOperation {
-                        actions: vec![CreateAlias {
+            .block_on(toc.submit_collection_operation(
+                CollectionMetaOperations::ChangeAliases(ChangeAliasesOperation {
+                    actions: vec![CreateAlias {
                            collection_name: "test".to_string(),
                            alias_name: "test_alias".to_string(),
                        }
                        .into()],
-                    },
-                )),
-            )
+                }),
+                None,
+            ))
            .unwrap();

        handle
-            .block_on(
-                toc.submit_collection_operation(CollectionMetaOperations::ChangeAliases(
-                    ChangeAliasesOperation {
-                        actions: vec![
+            .block_on(toc.submit_collection_operation(
+                CollectionMetaOperations::ChangeAliases(ChangeAliasesOperation {
+                    actions: vec![
                            CreateAlias {
                                collection_name: "test".to_string(),
                                alias_name: "test_alias2".to_string(),
@@ -94,9 +91,9 @@ mod tests {
                            }
                            .into(),
                        ],
-                    },
-                )),
-            )
+                }),
+                None,
+            ))
            .unwrap();

        handle.block_on(toc.get_collection("test_alias3")).unwrap();
--- a/src/actix/api/collections_api.rs
+++ b/src/actix/api/collections_api.rs
@@ -34,7 +34,7 @@ async fn update_collections(
    operation: web::Json<CollectionMetaOperations>,
 ) -> impl Responder {
    let timing = Instant::now();
-    let response = toc.submit_collection_operation(operation.0).await;
+    let response = toc.submit_collection_operation(operation.0, None).await;
    process_response(response, timing)
 }

@@ -47,12 +47,13 @@ async fn create_collection(
    let timing = Instant::now();
    let name = path.into_inner();
    let response = toc
-        .submit_collection_operation(CollectionMetaOperations::CreateCollection(
-            CreateCollectionOperation {
+        .submit_collection_operation(
+            CollectionMetaOperations::CreateCollection(CreateCollectionOperation {
                collection_name: name,
                create_collection: operation.0,
-            },
-        ))
+            }),
+            None,
+        )
        .await;
    process_response(response, timing)
 }
@@ -66,12 +67,13 @@ async fn update_collection(
    let timing = Instant::now();
    let name = path.into_inner();
    let response = toc
-        .submit_collection_operation(CollectionMetaOperations::UpdateCollection(
-            UpdateCollectionOperation {
+        .submit_collection_operation(
+            CollectionMetaOperations::UpdateCollection(UpdateCollectionOperation {
                collection_name: name,
                update_collection: operation.0,
-            },
-        ))
+            }),
+            None,
+        )
        .await;
    process_response(response, timing)
 }
@@ -84,9 +86,10 @@ async fn delete_collection(
    let timing = Instant::now();
    let name = path.into_inner();
    let response = toc
-        .submit_collection_operation(CollectionMetaOperations::DeleteCollection(
-            DeleteCollectionOperation(name),
-        ))
+        .submit_collection_operation(
+            CollectionMetaOperations::DeleteCollection(DeleteCollectionOperation(name)),
+            None,
+        )
        .await;
    process_response(response, timing)
 }
@@ -98,7 +101,7 @@ async fn update_aliases(
 ) -> impl Responder {
    let timing = Instant::now();
    let response = toc
-        .submit_collection_operation(CollectionMetaOperations::ChangeAliases(operation.0))
+        .submit_collection_operation(CollectionMetaOperations::ChangeAliases(operation.0), None)
        .await;
    process_response(response, timing)
 }
--- a/src/consensus.rs
+++ b/src/consensus.rs
@@ -187,7 +187,6 @@ mod tests {
        env_logger::init();
        let runtime = crate::create_search_runtime(settings.storage.performance.max_search_threads)
            .expect("Can't create runtime.");
-        let handle = runtime.handle().clone();
        let mut toc = TableOfContent::new(&settings.storage, runtime);
        let (propose_sender, propose_receiver) = std::sync::mpsc::channel();
        toc.with_propose_sender(propose_sender);
@@ -215,7 +214,10 @@ mod tests {
        assert_eq!(toc_arc.all_collections_sync().len(), 0);

        // When
-        handle
+
+        // New runtime is used as timers need to be enabled.
+        tokio::runtime::Runtime::new()
+            .unwrap()
            .block_on(toc_arc.submit_collection_operation(
                CollectionMetaOperations::CreateCollection(CreateCollectionOperation {
                    collection_name: "test".to_string(),
@@ -228,9 +230,9 @@ mod tests {
                        shard_number: 1,
                    },
                }),
+                None,
            ))
            .unwrap();
-        thread::sleep(Duration::from_secs(5));

        // Then
        assert_eq!(toc_arc.hard_state().unwrap().commit, 2);
--- a/src/tonic/api/collections_api.rs
+++ b/src/tonic/api/collections_api.rs
@@ -65,7 +65,7 @@ impl Collections for CollectionsService {
        let timing = Instant::now();
        let result = self
            .toc
-            .submit_collection_operation(operations)
+            .submit_collection_operation(operations, None)
            .await
            .map_err(error_to_status)?;

@@ -84,7 +84,7 @@ impl Collections for CollectionsService {
        let timing = Instant::now();
        let result = self
            .toc
-            .submit_collection_operation(operations)
+            .submit_collection_operation(operations, None)
            .await
            .map_err(error_to_status)?;

@@ -103,7 +103,7 @@ impl Collections for CollectionsService {
        let timing = Instant::now();
        let result = self
            .toc
-            .submit_collection_operation(operations)
+            .submit_collection_operation(operations, None)
            .await
            .map_err(error_to_status)?;

@@ -122,7 +122,7 @@ impl Collections for CollectionsService {
        let timing = Instant::now();
        let result = self
            .toc
-            .submit_collection_operation(operations)
+            .submit_collection_operation(operations, None)
            .await
            .map_err(error_to_status)?;