errors: introduce SchemaAgreementError

Few things happen here.

1. Introducing new error type SchemaAgreementTimeout
This is rather a rename of `SchemaVersionFetch` error type. We extend the type
by additional variants.

2. Removing ProtocolError::SchemaVersionFetch variant

3. Narrowing the error type of schema agreement methods
Thanks to that,  public API methods (e.g. Session::await_schema_agreement)
can return a self-contained error type providing more context than previously
returned ExecutionError.

4. Replacing ExecutionError::SchemaAgreementTimeout with SchemaAgreementError variant
SchemaAgreementError dependency is still needed, because of auto await schema agreement
configuration. By default, the driver awaits schema agreement on schema altering
queries.

5. Adjusting the example
Previously, the example would (unjustifiably) match against ExecutionError::RequestTimeout
error. Even before this PR, it should match against ExecutionError::SchemaAgreementTimeout.
Thanks to this commit, we were able to spot and fix this issue.

Author: muzarski

examples/schema_agreement.rs

@@ -2,7 +2,7 @@ use anyhow::{bail, Result};
 use futures::TryStreamExt as _;
 use scylla::client::session::Session;
 use scylla::client::session_builder::SessionBuilder;
-use scylla::errors::ExecutionError;
+use scylla::errors::SchemaAgreementError;
 use std::env;
 use std::time::Duration;
 
@@ -27,7 +27,9 @@ async fn main() -> Result<()> {
 
     match session.await_schema_agreement().await {
         Ok(_schema_version) => println!("Schema is in agreement in time"),
-        Err(ExecutionError::RequestTimeout(_)) => println!("Schema is NOT in agreement in time"),
+        Err(SchemaAgreementError::Timeout(_)) => {
+            println!("Schema is NOT in agreement in time")
+        }
         Err(err) => bail!(err),
     };
     session

scylla/src/client/session.rs

@@ -15,7 +15,7 @@ use crate::cluster::node::{InternalKnownNode, KnownNode, NodeRef};
 use crate::cluster::{Cluster, ClusterNeatDebug, ClusterState};
 use crate::errors::{
     BadQuery, ExecutionError, MetadataError, NewSessionError, PagerExecutionError, PrepareError,
-    RequestAttemptError, RequestError, TracingError, UseKeyspaceError,
+    RequestAttemptError, RequestError, SchemaAgreementError, TracingError, UseKeyspaceError,
 };
 use crate::frame::response::result;
 #[cfg(feature = "ssl")]
@@ -1938,7 +1938,7 @@ impl Session {
         last_error.map(Result::Err)
     }
 
-    async fn await_schema_agreement_indefinitely(&self) -> Result<Uuid, ExecutionError> {
+    async fn await_schema_agreement_indefinitely(&self) -> Result<Uuid, SchemaAgreementError> {
         loop {
             tokio::time::sleep(self.schema_agreement_interval).await;
             if let Some(agreed_version) = self.check_schema_agreement().await? {
@@ -1947,18 +1947,18 @@ impl Session {
         }
     }
 
-    pub async fn await_schema_agreement(&self) -> Result<Uuid, ExecutionError> {
+    pub async fn await_schema_agreement(&self) -> Result<Uuid, SchemaAgreementError> {
         timeout(
             self.schema_agreement_timeout,
             self.await_schema_agreement_indefinitely(),
         )
         .await
-        .unwrap_or(Err(ExecutionError::SchemaAgreementTimeout(
+        .unwrap_or(Err(SchemaAgreementError::Timeout(
             self.schema_agreement_timeout,
         )))
     }
 
-    pub async fn check_schema_agreement(&self) -> Result<Option<Uuid>, ExecutionError> {
+    pub async fn check_schema_agreement(&self) -> Result<Option<Uuid>, SchemaAgreementError> {
         let cluster_state = self.get_cluster_state();
         let connections_iter = cluster_state.iter_working_connections()?;
 

scylla/src/errors.rs

@@ -80,13 +80,13 @@ pub enum ExecutionError {
     )]
     RequestTimeout(std::time::Duration),
 
-    /// Schema agreement timed out.
-    #[error("Schema agreement exceeded {}ms", std::time::Duration::as_millis(.0))]
-    SchemaAgreementTimeout(std::time::Duration),
-
     /// 'USE KEYSPACE <>' request failed.
     #[error("'USE KEYSPACE <>' request failed: {0}")]
     UseKeyspaceError(#[from] UseKeyspaceError),
+
+    /// Failed to await automatic schema agreement.
+    #[error("Failed to await schema agreement: {0}")]
+    SchemaAgreementError(#[from] SchemaAgreementError),
 }
 
 impl From<SerializationError> for ExecutionError {
@@ -162,10 +162,6 @@ pub enum NewSessionError {
 #[derive(Error, Debug, Clone)]
 #[non_exhaustive]
 pub enum ProtocolError {
-    /// A protocol error appeared during schema version fetch.
-    #[error("Schema version fetch protocol error: {0}")]
-    SchemaVersionFetch(#[from] SchemaVersionFetchError),
-
     /// Unable extract a partition key based on prepared statement's metadata.
     #[error("Unable extract a partition key based on prepared statement's metadata")]
     PartitionKeyExtraction,
@@ -198,17 +194,33 @@ pub enum UseKeyspaceError {
     RequestTimeout(std::time::Duration),
 }
 
-/// A protocol error that occurred during schema version fetch.
+/// An error that occurred when awating schema agreement.
 #[derive(Error, Debug, Clone)]
 #[non_exhaustive]
-pub enum SchemaVersionFetchError {
+pub enum SchemaAgreementError {
+    /// Failed to find a node with working connection pool.
+    #[error("Failed to find a node with working connection pool: {0}")]
+    ConnectionPoolError(#[from] ConnectionPoolError),
+
+    /// Failed to execute schema version query on one of the connections.
+    ///
+    /// The driver attempts to fetch schema version on all connections in the pool (for all nodes).
+    /// It expects all of the requests to succeed. If at least one request fails, schema version
+    /// fetch is considered failed. This variant contains an error from one of the failing request attempts.
+    #[error("Failed to execute schema version query: {0}")]
+    RequestError(#[from] RequestAttemptError),
+
     /// Failed to convert schema version query result into rows result.
     #[error("Failed to convert schema version query result into rows result: {0}")]
     TracesEventsIntoRowsResultError(IntoRowsResultError),
 
     /// Failed to deserialize a single row from schema version query response.
     #[error(transparent)]
     SingleRowError(SingleRowError),
+
+    /// Schema agreement timed out.
+    #[error("Schema agreement exceeded {}ms", std::time::Duration::as_millis(.0))]
+    Timeout(std::time::Duration),
 }
 
 /// An error that occurred during tracing info fetch.

scylla/src/network/connection.rs

@@ -10,8 +10,8 @@ use crate::cluster::NodeAddr;
 use crate::errors::{
     BadKeyspaceName, BrokenConnectionError, BrokenConnectionErrorKind, ConnectionError,
     ConnectionSetupRequestError, ConnectionSetupRequestErrorKind, CqlEventHandlingError, DbError,
-    ExecutionError, InternalRequestError, ProtocolError, RequestAttemptError, ResponseParseError,
-    SchemaVersionFetchError, TranslationError, UseKeyspaceError,
+    InternalRequestError, RequestAttemptError, ResponseParseError, SchemaAgreementError,
+    TranslationError, UseKeyspaceError,
 };
 use crate::frame::protocol_features::ProtocolFeatures;
 use crate::frame::{
@@ -1261,20 +1261,14 @@ impl Connection {
         }
     }
 
-    pub(crate) async fn fetch_schema_version(&self) -> Result<Uuid, ExecutionError> {
+    pub(crate) async fn fetch_schema_version(&self) -> Result<Uuid, SchemaAgreementError> {
         let (version_id,) = self
             .query_unpaged(LOCAL_VERSION)
             .await?
             .into_rows_result()
-            .map_err(|err| {
-                ExecutionError::ProtocolError(ProtocolError::SchemaVersionFetch(
-                    SchemaVersionFetchError::TracesEventsIntoRowsResultError(err),
-                ))
-            })?
+            .map_err(SchemaAgreementError::TracesEventsIntoRowsResultError)?
             .single_row::<(Uuid,)>()
-            .map_err(|err| {
-                ProtocolError::SchemaVersionFetch(SchemaVersionFetchError::SingleRowError(err))
-            })?;
+            .map_err(SchemaAgreementError::SingleRowError)?;
 
         Ok(version_id)
     }