Merge pull request #1080 from muzarski/protocol-error-refactor

errors: protocol error refactor

scylla/src/statement/prepared_statement.rs

@@ -18,7 +18,7 @@ use crate::frame::types::{Consistency, SerialConsistency};
 use crate::history::HistoryListener;
 use crate::retry_policy::RetryPolicy;
 use crate::routing::Token;
-use crate::transport::errors::{BadQuery, QueryError};
+use crate::transport::errors::{BadQuery, ProtocolError, QueryError};
 use crate::transport::execution_profile::ExecutionProfileHandle;
 use crate::transport::partitioner::{Partitioner, PartitionerHasher, PartitionerName};
 
@@ -241,7 +241,7 @@ impl PreparedStatement {
  self.extract_partition_key(serialized_values)
  .map_err(|err| match err {
  PartitionKeyExtractionError::NoPkIndexValue(_, _) => {
- QueryError::ProtocolError("No pk indexes - can't calculate token")
+ ProtocolError::PartitionKeyExtraction
  }
  })?;
  let token = partition_key

scylla/src/transport/connection.rs

@@ -46,9 +46,9 @@ use std::{
  net::{Ipv4Addr, Ipv6Addr},
 };
 
+use super::errors::{ProtocolError, UseKeyspaceProtocolError};
 use super::iterator::RowIterator;
 use super::locator::tablets::{RawTablet, TabletParsingError};
-use super::query_result::SingleRowTypedError;
 use super::session::AddressTranslator;
 use super::topology::{PeerEndpoint, UntranslatedEndpoint, UntranslatedPeer};
 use super::NodeAddr;
@@ -299,10 +299,11 @@ impl NonErrorQueryResponse {
  let (result, paging_state) = self.into_query_result_and_paging_state()?;
 
  if !paging_state.finished() {
- let error_msg = "Internal driver API misuse or a server bug: nonfinished paging state\
- would be discarded by `NonErrorQueryResponse::into_query_result`";
- error!(error_msg);
- return Err(QueryError::ProtocolError(error_msg));
+ error!(
+ "Internal driver API misuse or a server bug: nonfinished paging state\
+ would be discarded by `NonErrorQueryResponse::into_query_result`"
+ );
+ return Err(ProtocolError::NonfinishedPagingState.into());
  }
 
  Ok(result)
@@ -904,7 +905,11 @@ impl Connection {
  // Reprepared statement should keep its id - it's the md5 sum
  // of statement contents
  if reprepared.get_id() != previous_prepared.get_id() {
- Err(UserRequestError::RepreparedIdChanged)
+ Err(UserRequestError::RepreparedIdChanged {
+ statement: reprepare_query.contents,
+ expected_id: previous_prepared.get_id().clone().into(),
+ reprepared_id: reprepared.get_id().clone().into(),
+ })
  } else {
  Ok(())
  }
@@ -1284,17 +1289,16 @@ impl Connection {
  self.reprepare(p.get_statement(), p).await?;
  continue;
  } else {
- return Err(QueryError::ProtocolError(
- "The server returned a prepared statement Id that did not exist in the batch",
- ));
+ return Err(ProtocolError::RepreparedIdMissingInBatch.into());
  }
  }
  _ => Err(err.into()),
  },
  Response::Result(_) => Ok(query_response.into_query_result()?),
- _ => Err(QueryError::ProtocolError(
- "BATCH: Unexpected server response",
- )),
+ _ => Err(ProtocolError::UnexpectedResponse(
+ query_response.response.to_response_kind(),
+ )
+ .into()),
  };
  }
  }
@@ -1359,23 +1363,40 @@ impl Connection {
  };
 
  let query_response = self.query_raw_unpaged(&query, PagingState::start()).await?;
+ Self::verify_use_keyspace_result(keyspace_name, query_response)
+ }
 
+ fn verify_use_keyspace_result(
+ keyspace_name: &VerifiedKeyspaceName,
+ query_response: QueryResponse,
+ ) -> Result<(), QueryError> {
  match query_response.response {
  Response::Result(result::Result::SetKeyspace(set_keyspace)) => {
- if set_keyspace.keyspace_name.to_lowercase()
- != keyspace_name.as_str().to_lowercase()
+ if !set_keyspace
+ .keyspace_name
+ .eq_ignore_ascii_case(keyspace_name.as_str())
  {
- return Err(QueryError::ProtocolError(
- "USE <keyspace_name> returned response with different keyspace name",
- ));
+ let expected_keyspace_name_lowercase = keyspace_name.as_str().to_lowercase();
+ let result_keyspace_name_lowercase = set_keyspace.keyspace_name.to_lowercase();
+
+ return Err(ProtocolError::UseKeyspace(
+ UseKeyspaceProtocolError::KeyspaceNameMismatch {
+ expected_keyspace_name_lowercase,
+ result_keyspace_name_lowercase,
+ },
+ )
+ .into());
  }
 
  Ok(())
  }
  Response::Error(err) => Err(err.into()),
- _ => Err(QueryError::ProtocolError(
- "USE <keyspace_name> returned unexpected response",
- )),
+ _ => Err(
+ ProtocolError::UseKeyspace(UseKeyspaceProtocolError::UnexpectedResponse(
+ query_response.response.to_response_kind(),
+ ))
+ .into(),
+ ),
  }
  }
 
@@ -1425,17 +1446,7 @@ impl Connection {
  .query_unpaged(LOCAL_VERSION)
  .await?
  .single_row_typed()
- .map_err(|err| match err {
- SingleRowTypedError::RowsExpected(_) => {
- QueryError::ProtocolError("Version query returned not rows")
- }
- SingleRowTypedError::BadNumberOfRows(_) => {
- QueryError::ProtocolError("system.local query returned a wrong number of rows")
- }
- SingleRowTypedError::FromRowError(_) => {
- QueryError::ProtocolError("Row is not uuid type as it should be")
- }
- })?;
+ .map_err(ProtocolError::SchemaVersionFetch)?;
  Ok(version_id)
  }
 

scylla/src/transport/downgrading_consistency_retry_policy.rs

@@ -185,7 +185,9 @@ mod tests {
  use bytes::Bytes;
 
  use crate::test_utils::setup_tracing;
- use crate::transport::errors::{BadQuery, BrokenConnectionErrorKind, ConnectionPoolError};
+ use crate::transport::errors::{
+ BadQuery, BrokenConnectionErrorKind, ConnectionPoolError, ProtocolError,
+ };
 
  use super::*;
 
@@ -284,7 +286,7 @@ mod tests {
  cl,
  );
  downgrading_consistency_policy_assert_never_retries(
- QueryError::ProtocolError("test"),
+ ProtocolError::NonfinishedPagingState.into(),
  cl,
  );
  }