Repository: krojew/cdrs-tokio (language: rust)

Issue:
Add non exhaustive to proto enums
We should add non exhaustive to enums which are likely to change upstream. 


Discussion / hints:

Fix patch:
diff --git a/cassandra-protocol/src/consistency.rs b/cassandra-protocol/src/consistency.rs
index f29e37a..f70f470 100644
--- a/cassandra-protocol/src/consistency.rs
+++ b/cassandra-protocol/src/consistency.rs
@@ -13,6 +13,7 @@ use std::str::FromStr;
 /// To find more details about each consistency level please refer to the following documentation:
 /// <https://docs.datastax.com/en/cql-oss/3.x/cql/cql_reference/cqlshConsistency.html>
 #[derive(Debug, PartialEq, Clone, Copy, Display, Ord, PartialOrd, Eq, Hash)]
+#[non_exhaustive]
 pub enum Consistency {
     /// Closest replica, as determined by the snitch.
     /// If all replica nodes are down, write succeeds after a hinted handoff.
diff --git a/cassandra-protocol/src/error.rs b/cassandra-protocol/src/error.rs
index b437ce8..d1179bb 100644
--- a/cassandra-protocol/src/error.rs
+++ b/cassandra-protocol/src/error.rs
@@ -19,6 +19,7 @@ pub type Result<T> = result::Result<T, Error>;
 /// can be raised by CDRS driver. `Server` error is an error which are ones returned by
 /// a Server via result error frames.
 #[derive(Debug, ThisError)]
+#[non_exhaustive]
 pub enum Error {
     /// Internal IO error.
     #[error("IO error: {0}")]
@@ -63,8 +64,8 @@ pub enum Error {
     #[error("Unexpected schema change target: {0}")]
     UnexpectedSchemaChangeTarget(String),
     /// Unexpected additional error info.
-    #[error("Unexpected additional error info: {0}")]
-    UnexpectedAdditionalErrorInfo(CInt),
+    #[error("Unexpected error code: {0}")]
+    UnexpectedErrorCode(CInt),
     /// Unexpected write type.
     #[error("Unexpected write type: {0}")]
     UnexpectedWriteType(String),
@@ -141,9 +142,7 @@ impl Clone for Error {
             Error::UnexpectedSchemaChangeTarget(value) => {
                 Error::UnexpectedSchemaChangeTarget(value.clone())
             }
-            Error::UnexpectedAdditionalErrorInfo(value) => {
-                Error::UnexpectedAdditionalErrorInfo(*value)
-            }
+            Error::UnexpectedErrorCode(value) => Error::UnexpectedErrorCode(*value),
             Error::UnexpectedWriteType(value) => Error::UnexpectedWriteType(value.clone()),
             Error::NonRequestOpcode(value) => Error::NonRequestOpcode(*value),
             Error::NonResponseOpcode(value) => Error::NonResponseOpcode(*value),
diff --git a/cassandra-protocol/src/frame.rs b/cassandra-protocol/src/frame.rs
index 231db41..f97fc9b 100644
--- a/cassandra-protocol/src/frame.rs
+++ b/cassandra-protocol/src/frame.rs
@@ -314,6 +314,7 @@ impl Envelope {
 }
 
 #[derive(Debug, Error)]
+#[non_exhaustive]
 pub enum CheckEnvelopeSizeError {
     #[error("Not enough bytes!")]
     NotEnoughBytes,
@@ -324,6 +325,7 @@ pub enum CheckEnvelopeSizeError {
 }
 
 #[derive(Debug, Error)]
+#[non_exhaustive]
 pub enum ParseEnvelopeError {
     /// There are not enough bytes to parse a single envelope, [`Envelope::from_buffer`] should be recalled when it is possible that there are more bytes.
     #[error("Not enough bytes!")]
@@ -343,6 +345,7 @@ pub enum ParseEnvelopeError {
 
 /// Protocol version.
 #[derive(Debug, PartialEq, Copy, Clone, Ord, PartialOrd, Eq, Hash, Display)]
+#[non_exhaustive]
 pub enum Version {
     V3,
     V4,
@@ -427,6 +430,7 @@ impl Flags {
 }
 
 #[derive(Debug, PartialEq, Copy, Clone, Ord, PartialOrd, Eq, Hash, Display)]
+#[non_exhaustive]
 pub enum Opcode {
     Error,
     Startup,
diff --git a/cassandra-protocol/src/frame/events.rs b/cassandra-protocol/src/frame/events.rs
index 11b760c..3a76c01 100644
--- a/cassandra-protocol/src/frame/events.rs
+++ b/cassandra-protocol/src/frame/events.rs
@@ -37,6 +37,7 @@ const AGGREGATE: &str = "AGGREGATE";
 /// about a concrete change. It may be useful for subscription
 /// when you need only string representation of an event.
 #[derive(Debug, PartialEq, Copy, Clone, Ord, PartialOrd, Eq, Hash)]
+#[non_exhaustive]
 pub enum SimpleServerEvent {
     TopologyChange,
     StatusChange,
@@ -94,6 +95,7 @@ impl PartialEq<ServerEvent> for SimpleServerEvent {
 
 /// Full server event that contains all details about a concrete change.
 #[derive(Debug, Clone, Ord, PartialOrd, Eq, PartialEq, Hash)]
+#[non_exhaustive]
 pub enum ServerEvent {
     /// Events related to change in the cluster topology
     TopologyChange(TopologyChange),
@@ -171,6 +173,7 @@ impl FromCursor for TopologyChange {
 }
 
 #[derive(Debug, Copy, Clone, PartialEq, Ord, PartialOrd, Eq, Hash, Display)]
+#[non_exhaustive]
 pub enum TopologyChangeType {
     NewNode,
     RemovedNode,
@@ -223,6 +226,7 @@ impl FromCursor for StatusChange {
 }
 
 #[derive(Debug, Copy, Clone, PartialEq, Eq, Hash, PartialOrd, Ord, Display)]
+#[non_exhaustive]
 pub enum StatusChangeType {
     Up,
     Down,
@@ -282,6 +286,7 @@ impl FromCursor for SchemaChange {
 
 /// Represents type of changes.
 #[derive(Debug, Copy, Clone, PartialEq, Eq, Hash, PartialOrd, Ord, Display)]
+#[non_exhaustive]
 pub enum SchemaChangeType {
     Created,
     Updated,
@@ -314,6 +319,7 @@ impl FromCursor for SchemaChangeType {
 
 /// Refers to a target of changes were made.
 #[derive(Debug, Copy, Clone, PartialEq, Eq, Hash, PartialOrd, Ord, Display)]
+#[non_exhaustive]
 pub enum SchemaChangeTarget {
     Keyspace,
     Table,
@@ -352,6 +358,7 @@ impl FromCursor for SchemaChangeTarget {
 
 /// Information about changes made.
 #[derive(Debug, Clone, Ord, PartialOrd, Eq, PartialEq, Hash)]
+#[non_exhaustive]
 pub enum SchemaChangeOptions {
     /// Changes related to keyspaces. Contains keyspace name.
     Keyspace(String),
diff --git a/cassandra-protocol/src/frame/message_batch.rs b/cassandra-protocol/src/frame/message_batch.rs
index e61f5e9..28ccb70 100644
--- a/cassandra-protocol/src/frame/message_batch.rs
+++ b/cassandra-protocol/src/frame/message_batch.rs
@@ -130,6 +130,7 @@ impl FromCursor for BodyReqBatch {
 
 /// Batch type
 #[derive(Debug, Clone, Copy, PartialEq, Ord, PartialOrd, Eq, Hash, Display)]
+#[non_exhaustive]
 pub enum BatchType {
     /// The batch will be "logged". This is equivalent to a
     /// normal CQL3 batch statement.
diff --git a/cassandra-protocol/src/frame/message_error.rs b/cassandra-protocol/src/frame/message_error.rs
index da62991..6e2ca27 100644
--- a/cassandra-protocol/src/frame/message_error.rs
+++ b/cassandra-protocol/src/frame/message_error.rs
@@ -16,19 +16,17 @@ use std::net::SocketAddr;
 /// it could contain additional information represented by `additional_info` property.
 #[derive(Debug, PartialEq, Eq, Clone)]
 pub struct ErrorBody {
-    /// `CInt` that points to a type of error.
-    pub error_code: CInt,
-    /// Error message string.
+    /// Error message.
     pub message: String,
-    /// Additional information.
-    pub additional_info: AdditionalErrorInfo,
+    /// The type of error, possibly including type specific additional information.
+    pub ty: ErrorType,
 }
 
 impl Serialize for ErrorBody {
     fn serialize(&self, cursor: &mut Cursor<&mut Vec<u8>>, version: Version) {
-        self.error_code.serialize(cursor, version);
+        self.ty.to_error_code().serialize(cursor, version);
         serialize_str(cursor, &self.message, version);
-        self.additional_info.serialize(cursor, version);
+        self.ty.serialize(cursor, version);
     }
 }
 
@@ -36,20 +34,16 @@ impl FromCursor for ErrorBody {
     fn from_cursor(cursor: &mut Cursor<&[u8]>, version: Version) -> error::Result<ErrorBody> {
         let error_code = CInt::from_cursor(cursor, version)?;
         let message = from_cursor_str(cursor)?.to_string();
-        let additional_info =
-            AdditionalErrorInfo::from_cursor_with_code(cursor, error_code, version)?;
+        let ty = ErrorType::from_cursor_with_code(cursor, error_code, version)?;
 
-        Ok(ErrorBody {
-            error_code,
-            message,
-            additional_info,
-        })
+        Ok(ErrorBody { message, ty })
     }
 }
 
 /// Protocol-dependent failure information. V5 contains a map of endpoint->code entries, while
 /// previous versions contain only error count.
 #[derive(Debug, PartialEq, Eq, Clone)]
+#[non_exhaustive]
 pub enum FailureInfo {
     /// Represents the number of nodes that experience a failure while executing the request.
     NumFailures(CInt),
@@ -97,7 +91,8 @@ impl FromCursor for FailureInfo {
 /// Additional error info in accordance to
 /// [Cassandra protocol v4](<https://github.com/apache/cassandra/blob/trunk/doc/native_protocol_v4.spec>).
 #[derive(Debug, PartialEq, Eq, Clone)]
-pub enum AdditionalErrorInfo {
+#[non_exhaustive]
+pub enum ErrorType {
     Server,
     Protocol,
     Authentication,
@@ -118,72 +113,77 @@ pub enum AdditionalErrorInfo {
     Unprepared(UnpreparedError),
 }
 
-impl Serialize for AdditionalErrorInfo {
+impl Serialize for ErrorType {
     fn serialize(&self, cursor: &mut Cursor<&mut Vec<u8>>, version: Version) {
         match self {
-            AdditionalErrorInfo::Unavailable(unavailable) => unavailable.serialize(cursor, version),
-            AdditionalErrorInfo::WriteTimeout(write_timeout) => {
-                write_timeout.serialize(cursor, version)
-            }
-            AdditionalErrorInfo::ReadTimeout(read_timeout) => {
-                read_timeout.serialize(cursor, version)
-            }
-            AdditionalErrorInfo::ReadFailure(read_failure) => {
-                read_failure.serialize(cursor, version)
-            }
-            AdditionalErrorInfo::FunctionFailure(function_failure) => {
+            ErrorType::Unavailable(unavailable) => unavailable.serialize(cursor, version),
+            ErrorType::WriteTimeout(write_timeout) => write_timeout.serialize(cursor, version),
+            ErrorType::ReadTimeout(read_timeout) => read_timeout.serialize(cursor, version),
+            ErrorType::ReadFailure(read_failure) => read_failure.serialize(cursor, version),
+            ErrorType::FunctionFailure(function_failure) => {
                 function_failure.serialize(cursor, version)
             }
-            AdditionalErrorInfo::WriteFailure(write_failure) => {
-                write_failure.serialize(cursor, version)
-            }
-            AdditionalErrorInfo::AlreadyExists(already_exists) => {
-                already_exists.serialize(cursor, version)
-            }
-            AdditionalErrorInfo::Unprepared(unprepared) => unprepared.serialize(cursor, version),
+            ErrorType::WriteFailure(write_failure) => write_failure.serialize(cursor, version),
+            ErrorType::AlreadyExists(already_exists) => already_exists.serialize(cursor, version),
+            ErrorType::Unprepared(unprepared) => unprepared.serialize(cursor, version),
             _ => {}
         }
     }
 }
 
-impl AdditionalErrorInfo {
+impl ErrorType {
     pub fn from_cursor_with_code(
         cursor: &mut Cursor<&[u8]>,
         error_code: CInt,
         version: Version,
-    ) -> error::Result<AdditionalErrorInfo> {
+    ) -> error::Result<ErrorType> {
         match error_code {
-            0x0000 => Ok(AdditionalErrorInfo::Server),
-            0x000A => Ok(AdditionalErrorInfo::Protocol),
-            0x0100 => Ok(AdditionalErrorInfo::Authentication),
-            0x1000 => {
-                UnavailableError::from_cursor(cursor, version).map(AdditionalErrorInfo::Unavailable)
-            }
-            0x1001 => Ok(AdditionalErrorInfo::Overloaded),
-            0x1002 => Ok(AdditionalErrorInfo::IsBootstrapping),
-            0x1003 => Ok(AdditionalErrorInfo::Truncate),
-            0x1100 => WriteTimeoutError::from_cursor(cursor, version)
-                .map(AdditionalErrorInfo::WriteTimeout),
-            0x1200 => {
-                ReadTimeoutError::from_cursor(cursor, version).map(AdditionalErrorInfo::ReadTimeout)
-            }
-            0x1300 => {
-                ReadFailureError::from_cursor(cursor, version).map(AdditionalErrorInfo::ReadFailure)
+            0x0000 => Ok(ErrorType::Server),
+            0x000A => Ok(ErrorType::Protocol),
+            0x0100 => Ok(ErrorType::Authentication),
+            0x1000 => UnavailableError::from_cursor(cursor, version).map(ErrorType::Unavailable),
+            0x1001 => Ok(ErrorType::Overloaded),
+            0x1002 => Ok(ErrorType::IsBootstrapping),
+            0x1003 => Ok(ErrorType::Truncate),
+            0x1100 => WriteTimeoutError::from_cursor(cursor, version).map(ErrorType::WriteTimeout),
+            0x1200 => ReadTimeoutError::from_cursor(cursor, version).map(ErrorType::ReadTimeout),
+            0x1300 => ReadFailureError::from_cursor(cursor, version).map(ErrorType::ReadFailure),
+            0x1400 => {
+                FunctionFailureError::from_cursor(cursor, version).map(ErrorType::FunctionFailure)
             }
-            0x1400 => FunctionFailureError::from_cursor(cursor, version)
-                .map(AdditionalErrorInfo::FunctionFailure),
-            0x1500 => WriteFailureError::from_cursor(cursor, version)
-                .map(AdditionalErrorInfo::WriteFailure),
-            0x2000 => Ok(AdditionalErrorInfo::Syntax),
-            0x2100 => Ok(AdditionalErrorInfo::Unauthorized),
-            0x2200 => Ok(AdditionalErrorInfo::Invalid),
-            0x2300 => Ok(AdditionalErrorInfo::Config),
-            0x2400 => AlreadyExistsError::from_cursor(cursor, version)
-                .map(AdditionalErrorInfo::AlreadyExists),
-            0x2500 => {
-                UnpreparedError::from_cursor(cursor, version).map(AdditionalErrorInfo::Unprepared)
+            0x1500 => WriteFailureError::from_cursor(cursor, version).map(ErrorType::WriteFailure),
+            0x2000 => Ok(ErrorType::Syntax),
+            0x2100 => Ok(ErrorType::Unauthorized),
+            0x2200 => Ok(ErrorType::Invalid),
+            0x2300 => Ok(ErrorType::Config),
+            0x2400 => {
+                AlreadyExistsError::from_cursor(cursor, version).map(ErrorType::AlreadyExists)
             }
-            _ => Err(Error::UnexpectedAdditionalErrorInfo(error_code)),
+            0x2500 => UnpreparedError::from_cursor(cursor, version).map(ErrorType::Unprepared),
+            _ => Err(Error::UnexpectedErrorCode(error_code)),
+        }
+    }
+
+    pub fn to_error_code(&self) -> CInt {
+        match self {
+            ErrorType::Server => 0x0000,
+            ErrorType::Protocol => 0x000A,
+            ErrorType::Authentication => 0x0100,
+            ErrorType::Unavailable(_) => 0x1000,
+            ErrorType::Overloaded => 0x1001,
+            ErrorType::IsBootstrapping => 0x1002,
+            ErrorType::Truncate => 0x1003,
+            ErrorType::WriteTimeout(_) => 0x1100,
+            ErrorType::ReadTimeout(_) => 0x1200,
+            ErrorType::ReadFailure(_) => 0x1300,
+            ErrorType::FunctionFailure(_) => 0x1400,
+            ErrorType::WriteFailure(_) => 0x1500,
+            ErrorType::Syntax => 0x2000,
+            ErrorType::Unauthorized => 0x2100,
+            ErrorType::Invalid => 0x2200,
+            ErrorType::Config => 0x2300,
+            ErrorType::AlreadyExists(_) => 0x2400,
+            ErrorType::Unprepared(_) => 0x2500,
         }
     }
 }
@@ -226,7 +226,7 @@ impl FromCursor for UnavailableError {
 }
 
 /// Timeout exception during a write request.
-#[derive(Debug, PartialEq, Copy, Clone, Ord, PartialOrd, Eq, Hash)]
+#[derive(Debug, PartialEq, Clone, Ord, PartialOrd, Eq, Hash)]
 pub struct WriteTimeoutError {
     /// Consistency level of query.
     pub cl: Consistency,
@@ -234,8 +234,11 @@ pub struct WriteTimeoutError {
     pub received: CInt,
     /// `i32` representing the number of replicas whose acknowledgement is required to achieve `cl`.
     pub block_for: CInt,
-    /// Describes the type of the write that timed out
+    /// Describes the type of the write that timed out.
     pub write_type: WriteType,
+    /// The number of contentions occurred during the CAS operation. The field only presents when
+    /// the `write_type` is `Cas`.
+    pub contentions: Option<CIntShort>,
 }
 
 impl Serialize for WriteTimeoutError {
@@ -244,6 +247,10 @@ impl Serialize for WriteTimeoutError {
         self.received.serialize(cursor, version);
         self.block_for.serialize(cursor, version);
         self.write_type.serialize(cursor, version);
+
+        if let Some(contentions) = self.contentions {
+            contentions.serialize(cursor, version);
+        }
     }
 }
 
@@ -256,12 +263,18 @@ impl FromCursor for WriteTimeoutError {
         let received = CInt::from_cursor(cursor, version)?;
         let block_for = CInt::from_cursor(cursor, version)?;
         let write_type = WriteType::from_cursor(cursor, version)?;
+        let contentions = if write_type == WriteType::Cas {
+            Some(CIntShort::from_cursor(cursor, version)?)
+        } else {
+            None
+        };
 
         Ok(WriteTimeoutError {
             cl,
             received,
             block_for,
             write_type,
+            contentions,
         })
     }
 }
@@ -459,21 +472,31 @@ impl FromCursor for WriteFailureError {
 
 /// Describes the type of the write that failed.
 /// [Read more...](https://github.com/apache/cassandra/blob/trunk/doc/native_protocol_v4.spec#L1118)
-#[derive(Debug, Ord, PartialOrd, Eq, PartialEq, Hash, Copy, Clone, Display)]
+#[derive(Debug, Ord, PartialOrd, Eq, PartialEq, Hash, Clone, Display)]
+#[non_exhaustive]
 pub enum WriteType {
-    /// The write was a non-batched non-counter write
+    /// The write was a non-batched non-counter write.
     Simple,
-    /// The write was a (logged) batch write.
-    /// If this type is received, it means the batch log
-    /// has been successfully written
+    /// The write was a (logged) batch write. If this type is received, it means the batch log
+    /// has been successfully written.
     Batch,
     /// The write was an unlogged batch. No batch log write has been attempted.
     UnloggedBatch,
-    /// The write was a counter write (batched or not)
+    /// The write was a counter write (batched or not).
     Counter,
     /// The failure occurred during the write to the batch log when a (logged) batch
     /// write was requested.
     BatchLog,
+    /// The timeout occurred during the Compare And Set write/update.
+    Cas,
+    /// The timeout occurred when a write involves VIEW update and failure to acquire local view(MV)
+    /// lock for key within timeout.
+    View,
+    /// The timeout occurred when cdc_total_space is exceeded when doing a write to data tracked by
+    /// cdc.
+    Cdc,
+    /// Unknown write type.
+    Unknown(String),
 }
 
 impl Serialize for WriteType {
@@ -484,6 +507,10 @@ impl Serialize for WriteType {
             WriteType::UnloggedBatch => serialize_str(cursor, "UNLOGGED_BATCH", version),
             WriteType::Counter => serialize_str(cursor, "COUNTER", version),
             WriteType::BatchLog => serialize_str(cursor, "BATCH_LOG", version),
+            WriteType::Cas => serialize_str(cursor, "CAS", version),
+            WriteType::View => serialize_str(cursor, "VIEW", version),
+            WriteType::Cdc => serialize_str(cursor, "CDC", version),
+            WriteType::Unknown(write_type) => serialize_str(cursor, write_type, version),
         }
     }
 }
@@ -496,7 +523,10 @@ impl FromCursor for WriteType {
             "UNLOGGED_BATCH" => Ok(WriteType::UnloggedBatch),
             "COUNTER" => Ok(WriteType::Counter),
             "BATCH_LOG" => Ok(WriteType::BatchLog),
-            wt => Err(Error::UnexpectedWriteType(wt.into())),
+            "CAS" => Ok(WriteType::Cas),
+            "VIEW" => Ok(WriteType::View),
+            "CDC" => Ok(WriteType::Cdc),
+            wt => Ok(WriteType::Unknown(wt.into())),
         }
     }
 }
@@ -553,6 +583,7 @@ impl FromCursor for UnpreparedError {
     }
 }
 
+//noinspection DuplicatedCode
 #[cfg(test)]
 fn test_encode_decode(bytes: &[u8], expected: ErrorBody) {
     {
@@ -580,9 +611,8 @@ mod error_tests {
             0, 3, 102, 111, 111, // message - foo
         ];
         let expected = ErrorBody {
-            error_code: 0x0000,
             message: "foo".into(),
-            additional_info: AdditionalErrorInfo::Server,
+            ty: ErrorType::Server,
         };
         test_encode_decode(bytes, expected);
     }
@@ -594,9 +624,8 @@ mod error_tests {
             0, 3, 102, 111, 111, // message - foo
         ];
         let expected = ErrorBody {
-            error_code: 0x000A,
             message: "foo".into(),
-            additional_info: AdditionalErrorInfo::Protocol,
+            ty: ErrorType::Protocol,
         };
         test_encode_decode(bytes, expected);
     }
@@ -608,9 +637,8 @@ mod error_tests {
             0, 3, 102, 111, 111, // message - foo
         ];
         let expected = ErrorBody {
-            error_code: 0x0100,
             message: "foo".into(),
-            additional_info: AdditionalErrorInfo::Authentication,
+            ty: ErrorType::Authentication,
         };
         test_encode_decode(bytes, expected);
     }
@@ -627,9 +655,8 @@ mod error_tests {
             0, 0, 0, 1, // alive
         ];
         let expected = ErrorBody {
-            error_code: 0x1000,
             message: "foo".into(),
-            additional_info: AdditionalErrorInfo::Unavailable(UnavailableError {
+            ty: ErrorType::Unavailable(UnavailableError {
                 cl: Consistency::Any,
                 required: 1,
                 alive: 1,
@@ -645,9 +672,8 @@ mod error_tests {
             0, 3, 102, 111, 111, // message - foo
         ];
         let expected = ErrorBody {
-            error_code: 0x1001,
             message: "foo".into(),
-            additional_info: AdditionalErrorInfo::Overloaded,
+            ty: ErrorType::Overloaded,
         };
         test_encode_decode(bytes, expected);
     }
@@ -659,9 +685,8 @@ mod error_tests {
             0, 3, 102, 111, 111, // message - foo
         ];
         let expected = ErrorBody {
-            error_code: 0x1002,
             message: "foo".into(),
-            additional_info: AdditionalErrorInfo::IsBootstrapping,
+            ty: ErrorType::IsBootstrapping,
         };
         test_encode_decode(bytes, expected);
     }
@@ -673,9 +698,8 @@ mod error_tests {
             0, 3, 102, 111, 111, // message - foo
         ];
         let expected = ErrorBody {
-            error_code: 0x1003,
             message: "foo".into(),
-            additional_info: AdditionalErrorInfo::Truncate,
+            ty: ErrorType::Truncate,
         };
         test_encode_decode(bytes, expected);
     }
@@ -693,13 +717,13 @@ mod error_tests {
             0, 6, 83, 73, 77, 80, 76, 69, // Write type simple
         ];
         let expected = ErrorBody {
-            error_code: 0x1100,
             message: "foo".into(),
-            additional_info: AdditionalErrorInfo::WriteTimeout(WriteTimeoutError {
+            ty: ErrorType::WriteTimeout(WriteTimeoutError {
                 cl: Consistency::Any,
                 received: 1,
                 block_for: 1,
                 write_type: WriteType::Simple,
+                contentions: None,
             }),
         };
         test_encode_decode(bytes, expected);
@@ -718,9 +742,8 @@ mod error_tests {
             0, // data present
         ];
         let expected = ErrorBody {
-            error_code: 0x1200,
             message: "foo".into(),
-            additional_info: AdditionalErrorInfo::ReadTimeout(ReadTimeoutError {
+            ty: ErrorType::ReadTimeout(ReadTimeoutError {
                 cl: Consistency::Any,
                 received: 1,
                 block_for: 1,
@@ -744,9 +767,8 @@ mod error_tests {
             0, // data present
         ];
         let expected = ErrorBody {
-            error_code: 0x1300,
             message: "foo".into(),
-            additional_info: AdditionalErrorInfo::ReadFailure(ReadFailureError {
+            ty: ErrorType::ReadFailure(ReadFailureError {
                 cl: Consistency::Any,
                 received: 1,
                 block_for: 1,
@@ -764,9 +786,8 @@ mod error_tests {
             0, 3, 102, 111, 111, // message - foo
         ];
         let expected = ErrorBody {
-            error_code: 0x2000,
             message: "foo".into(),
-            additional_info: AdditionalErrorInfo::Syntax,
+            ty: ErrorType::Syntax,
         };
         test_encode_decode(bytes, expected);
     }
@@ -778,9 +799,8 @@ mod error_tests {
             0, 3, 102, 111, 111, // message - foo
         ];
         let expected = ErrorBody {
-            error_code: 0x2100,
             message: "foo".into(),
-            additional_info: AdditionalErrorInfo::Unauthorized,
+            ty: ErrorType::Unauthorized,
         };
         test_encode_decode(bytes, expected);
     }
@@ -792,9 +812,8 @@ mod error_tests {
             0, 3, 102, 111, 111, // message - foo
         ];
         let expected = ErrorBody {
-            error_code: 0x2200,
             message: "foo".into(),
-            additional_info: AdditionalErrorInfo::Invalid,
+            ty: ErrorType::Invalid,
         };
         test_encode_decode(bytes, expected);
     }
@@ -806,9 +825,8 @@ mod error_tests {
             0, 3, 102, 111, 111, // message - foo
         ];
         let expected = ErrorBody {
-            error_code: 0x2300,
             message: "foo".into(),
-            additional_info: AdditionalErrorInfo::Config,
+            ty: ErrorType::Config,
         };
         test_encode_decode(bytes, expected);
     }
diff --git a/cassandra-protocol/src/frame/message_query.rs b/cassandra-protocol/src/frame/message_query.rs
index f62a35e..ad40003 100644
--- a/cassandra-protocol/src/frame/message_query.rs
+++ b/cassandra-protocol/src/frame/message_query.rs
@@ -2,7 +2,7 @@ use crate::consistency::Consistency;
 use crate::error;
 use crate::frame::traits::FromCursor;
 use crate::frame::{Direction, Envelope, Flags, Opcode, Serialize, Version};
-use crate::query::{Query, QueryParams, QueryValues};
+use crate::query::{QueryParams, QueryValues};
 use crate::types::{from_cursor_str_long, serialize_str_long, CBytes, CInt, CLong, INT_LEN};
 use std::io::Cursor;
 
@@ -118,18 +118,18 @@ impl Envelope {
     }
 
     #[inline]
-    pub fn new_query(query: Query, flags: Flags, version: Version) -> Envelope {
+    pub fn new_query(query: BodyReqQuery, flags: Flags, version: Version) -> Envelope {
         Envelope::new_req_query(
             query.query,
-            query.params.consistency,
-            query.params.values,
-            query.params.with_names,
-            query.params.page_size,
-            query.params.paging_state,
-            query.params.serial_consistency,
-            query.params.timestamp,
-            query.params.keyspace,
-            query.params.now_in_seconds,
+            query.query_params.consistency,
+            query.query_params.values,
+            query.query_params.with_names,
+            query.query_params.page_size,
+            query.query_params.paging_state,
+            query.query_params.serial_consistency,
+            query.query_params.timestamp,
+            query.query_params.keyspace,
+            query.query_params.now_in_seconds,
             flags,
             version,
         )
diff --git a/cassandra-protocol/src/frame/message_request.rs b/cassandra-protocol/src/frame/message_request.rs
index dc539a3..b142c73 100644
--- a/cassandra-protocol/src/frame/message_request.rs
+++ b/cassandra-protocol/src/frame/message_request.rs
@@ -13,6 +13,7 @@ use crate::{error, Error};
 
 #[derive(Debug, PartialEq, Eq, Clone)]
 #[allow(clippy::large_enum_variant)]
+#[non_exhaustive]
 pub enum RequestBody {
     Startup(BodyReqStartup),
     Options(BodyReqOptions),
diff --git a/cassandra-protocol/src/frame/message_response.rs b/cassandra-protocol/src/frame/message_response.rs
index a36190b..cd24e14 100644
--- a/cassandra-protocol/src/frame/message_response.rs
+++ b/cassandra-protocol/src/frame/message_response.rs
@@ -14,6 +14,7 @@ use crate::types::rows::Row;
 use crate::{error, Error};
 
 #[derive(Debug, PartialEq, Eq, Clone)]
+#[non_exhaustive]
 pub enum ResponseBody {
     Error(ErrorBody),
     Ready,
diff --git a/cassandra-protocol/src/frame/message_result.rs b/cassandra-protocol/src/frame/message_result.rs
index f5bca8e..8a4d992 100644
--- a/cassandra-protocol/src/frame/message_result.rs
+++ b/cassandra-protocol/src/frame/message_result.rs
@@ -14,6 +14,7 @@ use std::io::{Cursor, Error as IoError, Read};
 
 /// `ResultKind` is enum which represents types of result.
 #[derive(Debug, Ord, PartialOrd, Eq, PartialEq, Copy, Clone, Hash, Display)]
+#[non_exhaustive]
 pub enum ResultKind {
     /// Void result.
     Void,
@@ -82,6 +83,7 @@ impl FromCursor for ResultKind {
 /// `ResponseBody` is a generalized enum that represents all types of responses. Each of enum
 /// option wraps related body type.
 #[derive(Debug, PartialEq, Ord, PartialOrd, Eq, Clone, Hash)]
+#[non_exhaustive]
 pub enum ResResultBody {
     /// Void response body. It's an empty struct.
     Void,
@@ -502,6 +504,7 @@ impl ColSpec {
 
 /// Cassandra data types which could be returned by a server.
 #[derive(Debug, Clone, Display, Copy, Ord, PartialOrd, Eq, PartialEq, Hash)]
+#[non_exhaustive]
 pub enum ColType {
     Custom,
     Ascii,
@@ -529,7 +532,6 @@ pub enum ColType {
     Set,
     Udt,
     Tuple,
-    Null,
 }
 
 impl TryFrom<CIntShort> for ColType {
@@ -607,7 +609,6 @@ impl Serialize for ColType {
             ColType::Set => 0x0022,
             ColType::Udt => 0x0030,
             ColType::Tuple => 0x0031,
-            _ => 0x6666,
         } as CIntShort)
             .serialize(cursor, version);
     }
@@ -680,6 +681,7 @@ impl FromCursor for ColTypeOption {
 
 /// Enum that represents all possible types of `value` of `ColTypeOption`.
 #[derive(Debug, Clone, PartialEq, Ord, PartialOrd, Eq, Hash)]
+#[non_exhaustive]
 pub enum ColTypeOptionValue {
     CString(String),
     ColType(ColType),
diff --git a/cassandra-protocol/src/query.rs b/cassandra-protocol/src/query.rs
index 0b7b644..4e02e4d 100644
--- a/cassandra-protocol/src/query.rs
+++ b/cassandra-protocol/src/query.rs
@@ -14,11 +14,3 @@ pub use crate::query::query_flags::QueryFlags;
 pub use crate::query::query_params::QueryParams;
 pub use crate::query::query_params_builder::QueryParamsBuilder;
 pub use crate::query::query_values::QueryValues;
-
-/// Structure that represents CQL query and parameters which will be applied during
-/// its execution
-#[derive(Debug, Default)]
-pub struct Query {
-    pub query: String,
-    pub params: QueryParams,
-}
diff --git a/cassandra-protocol/src/types.rs b/cassandra-protocol/src/types.rs
index 2bb8705..c499a6a 100644
--- a/cassandra-protocol/src/types.rs
+++ b/cassandra-protocol/src/types.rs
@@ -283,9 +283,9 @@ impl CBytes {
         CBytes { bytes: Some(bytes) }
     }
 
-    /// Creates Cassandra bytes that represent empty or null value
+    /// Creates Cassandra bytes that represent null value
     #[inline]
-    pub fn new_empty() -> CBytes {
+    pub fn new_null() -> CBytes {
         CBytes { bytes: None }
     }
 
@@ -301,7 +301,7 @@ impl CBytes {
     }
 
     #[inline]
-    pub fn is_empty(&self) -> bool {
+    pub fn is_null_or_empty(&self) -> bool {
         match &self.bytes {
             None => true,
             Some(bytes) => bytes.is_empty(),
diff --git a/cassandra-protocol/src/types/cassandra_type.rs b/cassandra-protocol/src/types/cassandra_type.rs
index 1493e90..b939699 100644
--- a/cassandra-protocol/src/types/cassandra_type.rs
+++ b/cassandra-protocol/src/types/cassandra_type.rs
@@ -9,6 +9,7 @@ use crate::frame::Version;
 use crate::types::CBytes;
 
 #[derive(Debug, PartialEq, Clone)]
+#[non_exhaustive]
 pub enum CassandraType {
     Ascii(String),
     Bigint(i64),
@@ -69,7 +70,6 @@ pub fn wrapper_fn(
         ColType::Set => &wrappers::set,
         ColType::Udt => &wrappers::udt,
         ColType::Tuple => &wrappers::tuple,
-        ColType::Null => &wrappers::null,
     }
 }
 
diff --git a/cassandra-protocol/src/types/data_serialization_types.rs b/cassandra-protocol/src/types/data_serialization_types.rs
index 4a2c2e4..475d0e3 100644
--- a/cassandra-protocol/src/types/data_serialization_types.rs
+++ b/cassandra-protocol/src/types/data_serialization_types.rs
@@ -234,7 +234,7 @@ pub fn decode_udt(bytes: &[u8], l: usize, version: Version) -> Result<Vec<CBytes
             .or_else(|err| match err {
                 error::Error::Io(io_err) => {
                     if io_err.kind() == io::ErrorKind::UnexpectedEof {
-                        Ok(CBytes::new_empty())
+                        Ok(CBytes::new_null())
                     } else {
                         Err(io_err.into())
                     }
diff --git a/cassandra-protocol/src/types/rows.rs b/cassandra-protocol/src/types/rows.rs
index e40c3d7..5a268a3 100644
--- a/cassandra-protocol/src/types/rows.rs
+++ b/cassandra-protocol/src/types/rows.rs
@@ -54,7 +54,7 @@ impl Row {
     pub fn is_empty(&self, index: usize) -> bool {
         self.row_content
             .get(index)
-            .map(|data| data.is_empty())
+            .map(|data| data.is_null_or_empty())
             .unwrap_or(false)
     }
 
diff --git a/cdrs-tokio/examples/crud_operations.rs b/cdrs-tokio/examples/crud_operations.rs
index 8677f79..be1e766 100644
--- a/cdrs-tokio/examples/crud_operations.rs
+++ b/cdrs-tokio/examples/crud_operations.rs
@@ -32,6 +32,7 @@ async fn main() {
     let mut session: CurrentSession =
         TcpSessionBuilder::new(RoundRobinLoadBalancingStrategy::new(), config)
             .build()
+            .await
             .unwrap();
 
     create_keyspace(&mut session).await;
diff --git a/cdrs-tokio/examples/insert_collection.rs b/cdrs-tokio/examples/insert_collection.rs
index a575752..7ccf5de 100644
--- a/cdrs-tokio/examples/insert_collection.rs
+++ b/cdrs-tokio/examples/insert_collection.rs
@@ -32,6 +32,7 @@ async fn main() {
     let mut session: CurrentSession =
         TcpSessionBuilder::new(RoundRobinLoadBalancingStrategy::new(), cluster_config)
             .build()
+            .await
             .unwrap();
 
     create_keyspace(&mut session).await;
diff --git a/cdrs-tokio/examples/multiple_thread.rs b/cdrs-tokio/examples/multiple_thread.rs
index ed9e388..e140133 100644
--- a/cdrs-tokio/examples/multiple_thread.rs
+++ b/cdrs-tokio/examples/multiple_thread.rs
@@ -24,8 +24,12 @@ async fn main() {
         .await
         .unwrap();
     let lb = RoundRobinLoadBalancingStrategy::new();
-    let session: Arc<CurrentSession> =
-        Arc::new(TcpSessionBuilder::new(lb, cluster_config).build().unwrap());
+    let session: Arc<CurrentSession> = Arc::new(
+        TcpSessionBuilder::new(lb, cluster_config)
+            .build()
+            .await
+            .unwrap(),
+    );
 
     create_keyspace(session.clone()).await;
     create_table(session.clone()).await;
diff --git a/cdrs-tokio/examples/paged_query.rs b/cdrs-tokio/examples/paged_query.rs
index f5656fd..eb946d2 100644
--- a/cdrs-tokio/examples/paged_query.rs
+++ b/cdrs-tokio/examples/paged_query.rs
@@ -50,7 +50,10 @@ async fn main() {
         .await
         .unwrap();
     let lb = RoundRobinLoadBalancingStrategy::new();
-    let session = TcpSessionBuilder::new(lb, cluster_config).build().unwrap();
+    let session = TcpSessionBuilder::new(lb, cluster_config)
+        .build()
+        .await
+        .unwrap();
 
     create_keyspace(&session).await;
     create_udt(&session).await;
@@ -196,11 +199,19 @@ async fn paged_with_values_list(session: &CurrentSession) {
     println!("Testing values 100 and 101");
     assert_amount_query_pager!(2);
     assert!(query_pager.has_more());
-    assert!(!query_pager.pager_state().cursor().unwrap().is_empty());
+    assert!(!query_pager
+        .pager_state()
+        .cursor()
+        .unwrap()
+        .is_null_or_empty());
     println!("Testing values 102 and 103");
     assert_amount_query_pager!(2);
     assert!(query_pager.has_more());
-    assert!(!query_pager.pager_state().cursor().unwrap().is_empty());
+    assert!(!query_pager
+        .pager_state()
+        .cursor()
+        .unwrap()
+        .is_null_or_empty());
     println!("Testing value 104");
     assert_amount_query_pager!(1);
     // Now no more rows should be queried
diff --git a/cdrs-tokio/examples/prepare_batch_execute.rs b/cdrs-tokio/examples/prepare_batch_execute.rs
index 19d054d..5c1d8ec 100644
--- a/cdrs-tokio/examples/prepare_batch_execute.rs
+++ b/cdrs-tokio/examples/prepare_batch_execute.rs
@@ -36,7 +36,10 @@ async fn main() {
         .await
         .unwrap();
     let lb = RoundRobinLoadBalancingStrategy::new();
-    let mut session = TcpSessionBuilder::new(lb, cluster_config).build().unwrap();
+    let mut session = TcpSessionBuilder::new(lb, cluster_config)
+        .build()
+        .await
+        .unwrap();
 
     create_keyspace(&mut session).await;
     create_table(&mut session).await;
diff --git a/cdrs-tokio/src/cluster/cluster_metadata_manager.rs b/cdrs-tokio/src/cluster/cluster_metadata_manager.rs
index 3d7004a..f7aa832 100644
--- a/cdrs-tokio/src/cluster/cluster_metadata_manager.rs
+++ b/cdrs-tokio/src/cluster/cluster_metadata_manager.rs
@@ -5,9 +5,10 @@ use cassandra_protocol::frame::events::{
     SchemaChangeOptions, SchemaChangeType, StatusChange, StatusChangeType, TopologyChange,
     TopologyChangeType,
 };
-use cassandra_protocol::frame::message_error::{AdditionalErrorInfo, ErrorBody};
+use cassandra_protocol::frame::message_error::{ErrorBody, ErrorType};
+use cassandra_protocol::frame::message_query::BodyReqQuery;
 use cassandra_protocol::frame::{Envelope, Flags, Version};
-use cassandra_protocol::query::{Query, QueryParams, QueryParamsBuilder, QueryValues};
+use cassandra_protocol::query::{QueryParams, QueryParamsBuilder, QueryValues};
 use cassandra_protocol::token::Murmur3Token;
 use cassandra_protocol::types::list::List;
 use cassandra_protocol::types::rows::Row;
@@ -78,9 +79,9 @@ async fn send_query_with_params<T: CdrsTransport>(
     version: Version,
     beta_protocol: bool,
 ) -> Result<Option<Vec<Row>>> {
-    let query = Query {
+    let query = BodyReqQuery {
         query: query.to_string(),
-        params: query_params,
+        query_params,
     };
 
     let flags = if beta_protocol {
@@ -369,6 +370,7 @@ impl<T: CdrsTransport + 'static, CM: ConnectionManager<T> + 'static> ClusterMeta
             ServerEvent::TopologyChange(event) => self.process_topology_event(event).await,
             ServerEvent::StatusChange(event) => self.process_status_event(event).await,
             ServerEvent::SchemaChange(event) => self.process_schema_event(event).await,
+            _ => warn!(?event, "Unrecognized event."),
         }
     }
 
@@ -381,6 +383,7 @@ impl<T: CdrsTransport + 'static, CM: ConnectionManager<T> + 'static> ClusterMeta
                 SchemaChangeType::Dropped => {
                     self.remove_keyspace(keyspace);
                 }
+                _ => warn!(?event, "Unrecognized schema event."),
             }
         }
     }
@@ -412,6 +415,7 @@ impl<T: CdrsTransport + 'static, CM: ConnectionManager<T> + 'static> ClusterMeta
                     );
                 }
             }
+            _ => warn!(?event, "Unrecognized topology change type."),
         }
     }
 
@@ -451,6 +455,7 @@ impl<T: CdrsTransport + 'static, CM: ConnectionManager<T> + 'static> ClusterMeta
                     debug!(broadcast_rpc_address = %event.addr, "Unknown node down.");
                 }
             }
+            _ => warn!(?event, "Unrecognized status event."),
         }
     }
 
@@ -706,7 +711,7 @@ impl<T: CdrsTransport + 'static, CM: ConnectionManager<T> + 'static> ClusterMeta
             Err(Error::Server {
                 body:
                     ErrorBody {
-                        additional_info: AdditionalErrorInfo::Invalid,
+                        ty: ErrorType::Invalid,
                         ..
                     },
                 ..
diff --git a/cdrs-tokio/src/cluster/control_connection.rs b/cdrs-tokio/src/cluster/control_connection.rs
index 977ec03..c79f2d7 100644
--- a/cdrs-tokio/src/cluster/control_connection.rs
+++ b/cdrs-tokio/src/cluster/control_connection.rs
@@ -38,11 +38,12 @@ impl<
         LB: LoadBalancingStrategy<T, CM> + Send + Sync,
     > ControlConnection<T, CM, LB>
 {
-    pub async fn run(self) {
+    pub async fn run(self, init_complete_sender: tokio::sync::oneshot::Sender<()>) {
         let (event_envelope_sender, event_envelope_receiver) = channel(EVENT_CHANNEL_CAPACITY);
         let (error_sender, mut error_receiver) = channel(1);
 
         Self::process_events(event_envelope_receiver, self.event_sender.clone());
+        let mut init_complete_sender = Some(init_complete_sender);
 
         'listen: loop {
             let current_connection = self
@@ -64,6 +65,9 @@ impl<
                 let result = current_connection
                     .write_envelope(&register_envelope, false)
                     .await;
+                if let Some(sender) = init_complete_sender.take() {
+                    sender.send(()).ok();
+                }
                 match result {
                     Ok(_) => {
                         let error = error_receiver.recv().await;
diff --git a/cdrs-tokio/src/cluster/session.rs b/cdrs-tokio/src/cluster/session.rs
index 4a45a65..cef901e 100644
--- a/cdrs-tokio/src/cluster/session.rs
+++ b/cdrs-tokio/src/cluster/session.rs
@@ -3,10 +3,12 @@ use cassandra_protocol::compression::Compression;
 use cassandra_protocol::consistency::Consistency;
 use cassandra_protocol::error;
 use cassandra_protocol::events::ServerEvent;
+use cassandra_protocol::frame::message_error::ErrorType;
+use cassandra_protocol::frame::message_query::BodyReqQuery;
 use cassandra_protocol::frame::message_response::ResponseBody;
 use cassandra_protocol::frame::message_result::{BodyResResultPrepared, TableSpec};
 use cassandra_protocol::frame::{Envelope, Flags, Serialize, Version};
-use cassandra_protocol::query::{PreparedQuery, Query, QueryBatch, QueryValues};
+use cassandra_protocol::query::{PreparedQuery, QueryBatch, QueryValues};
 use cassandra_protocol::token::Murmur3Token;
 use cassandra_protocol::types::value::Value;
 use cassandra_protocol::types::{CIntShort, SHORT_LEN};
@@ -41,6 +43,7 @@ use crate::cluster::{ClusterMetadata, ClusterMetadataManager, SessionContext};
 use crate::cluster::{GenericClusterConfig, KeyspaceHolder};
 use crate::cluster::{NodeTcpConfig, SessionPager};
 use crate::frame_encoding::{FrameEncodingFactory, ProtocolFrameEncodingFactory};
+use crate::future::BoxFuture;
 use crate::load_balancing::node_distance_evaluator::AllLocalNodeDistanceEvaluator;
 use crate::load_balancing::node_distance_evaluator::NodeDistanceEvaluator;
 use crate::load_balancing::{
@@ -293,7 +296,7 @@ impl<
 
         if let Err(error::Error::Server { body: error, addr }) = &result {
             // if query is unprepared
-            if error.error_code == 0x2500 {
+            if let ErrorType::Unprepared(_) = error.ty {
                 debug!("Re-preparing statement.");
 
                 // We need to send the prepare statement to the failing node.
@@ -544,9 +547,9 @@ impl<
             .as_ref()
             .map(|values| serialize_routing_key(values, self.version));
 
-        let query = Query {
+        let query = BodyReqQuery {
             query: query.to_string(),
-            params: parameters.query_params,
+            query_params: parameters.query_params,
         };
 
         let flags = prepare_flags(
@@ -750,7 +753,7 @@ impl<
     }
 
     #[allow(clippy::too_many_arguments)]
-    fn new(
+    async fn new(
         load_balancing: LB,
         keyspace_holder: Arc<KeyspaceHolder>,
         keyspace_receiver: watch::Receiver<Option<String>>,
@@ -764,7 +767,7 @@ impl<
         version: Version,
         connection_pool_config: ConnectionPoolConfig,
         beta_protocol: bool,
-    ) -> Self {
+    ) -> Result<Self, SessionBuildError> {
         let connection_pool_factory = Arc::new(ConnectionPoolFactory::new(
             connection_pool_config,
             version,
@@ -823,9 +826,13 @@ impl<
             version,
         );
 
-        let control_connection_handle = tokio::spawn(control_connection.run());
+        let (init_complete_sender, init_complete_receiver) = tokio::sync::oneshot::channel();
+        let control_connection_handle = tokio::spawn(control_connection.run(init_complete_sender));
+        if init_complete_receiver.await.is_err() {
+            return Err(SessionBuildError::SessionInitFailed);
+        }
 
-        Session {
+        Ok(Session {
             load_balancing,
             keyspace_holder,
             retry_policy,
@@ -836,7 +843,7 @@ impl<
             _transport: Default::default(),
             _connection_manager: Default::default(),
             version,
-        }
+        })
     }
 }
 
@@ -882,7 +889,7 @@ where
 {
     let (keyspace_holder, keyspace_receiver) = create_keyspace_holder();
     let connection_manager = config.create_manager(keyspace_holder.clone()).await?;
-    Ok(Session::new(
+    Session::new(
         load_balancing,
         keyspace_holder,
         keyspace_receiver,
@@ -896,7 +903,9 @@ where
         config.version(),
         config.connection_pool_config(),
         config.beta_protocol(),
-    ))
+    )
+    .await
+    .map_err(|e| error::Error::General(e.to_string()))
 }
 
 struct SessionConfig<
@@ -920,8 +929,8 @@ struct SessionConfig<
 }
 
 impl<
-        T: CdrsTransport,
-        CM: ConnectionManager<T>,
+        T: CdrsTransport + 'static,
+        CM: ConnectionManager<T> + 'static,
         LB: LoadBalancingStrategy<T, CM> + Send + Sync + 'static,
     > SessionConfig<T, CM, LB>
 {
@@ -943,7 +952,7 @@ impl<
         }
     }
 
-    fn into_session(
+    async fn into_session(
         self,
         keyspace_holder: Arc<KeyspaceHolder>,
         keyspace_receiver: watch::Receiver<Option<String>>,
@@ -951,7 +960,7 @@ impl<
         connection_manager: CM,
         version: Version,
         beta_protocol: bool,
-    ) -> Session<T, CM, LB> {
+    ) -> Result<Session<T, CM, LB>, SessionBuildError> {
         if let Some(keyspace) = self.keyspace {
             keyspace_holder.update_current_keyspace_without_notification(keyspace);
         }
@@ -971,6 +980,7 @@ impl<
             self.connection_pool_config,
             beta_protocol,
         )
+        .await
     }
 }
 
@@ -979,6 +989,8 @@ impl<
 pub enum SessionBuildError {
     #[error("Given compression type is not supported for selected protocol!")]
     CompressionTypeNotSupported,
+    #[error("Session control connection died before completing initialization")]
+    SessionInitFailed,
 }
 
 /// Builder for easy `Session` creation. Requires static `LoadBalancingStrategy`, but otherwise, other
@@ -1060,7 +1072,7 @@ pub trait SessionBuilder<
     fn with_beta_protocol(self, beta_protocol: bool) -> Self;
 
     /// Builds the resulting session.
-    fn build(self) -> Result<Session<T, CM, LB>, SessionBuildError>;
+    fn build(self) -> BoxFuture<'static, Result<Session<T, CM, LB>, SessionBuildError>>;
 }
 
 /// Builder for non-TLS sessions.
@@ -1161,31 +1173,45 @@ impl<LB: LoadBalancingStrategy<TransportTcp, TcpConnectionManager> + Send + Sync
         self
     }
 
-    fn build(self) -> Result<Session<TransportTcp, TcpConnectionManager, LB>, SessionBuildError> {
-        verify_compression_configuration(self.node_config.version, self.config.compression).map(
-            |()| {
-                let (keyspace_holder, keyspace_receiver) = create_keyspace_holder();
-                let connection_manager = TcpConnectionManager::new(
-                    self.node_config.authenticator_provider,
-                    keyspace_holder.clone(),
-                    self.config.reconnection_policy.clone(),
-                    self.frame_encoder_factory,
-                    self.config.compression,
-                    self.config.transport_buffer_size,
-                    self.config.tcp_nodelay,
-                    self.node_config.version,
-                );
+    fn build(
+        self,
+    ) -> BoxFuture<
+        'static,
+        Result<Session<TransportTcp, TcpConnectionManager, LB>, SessionBuildError>,
+    > {
+        async move {
+            match verify_compression_configuration(
+                self.node_config.version,
+                self.config.compression,
+            ) {
+                Ok(()) => {
+                    let (keyspace_holder, keyspace_receiver) = create_keyspace_holder();
+                    let connection_manager = TcpConnectionManager::new(
+                        self.node_config.authenticator_provider,
+                        keyspace_holder.clone(),
+                        self.config.reconnection_policy.clone(),
+                        self.frame_encoder_factory,
+                        self.config.compression,
+                        self.config.transport_buffer_size,
+                        self.config.tcp_nodelay,
+                        self.node_config.version,
+                    );
 
-                self.config.into_session(
-                    keyspace_holder,
-                    keyspace_receiver,
-                    self.node_config.contact_points,
-                    connection_manager,
-                    self.node_config.version,
-                    self.node_config.beta_protocol,
-                )
-            },
-        )
+                    self.config
+                        .into_session(
+                            keyspace_holder,
+                            keyspace_receiver,
+                            self.node_config.contact_points,
+                            connection_manager,
+                            self.node_config.version,
+                            self.node_config.beta_protocol,
+                        )
+                        .await
+                }
+                Err(err) => Err(err),
+            }
+        }
+        .boxed()
     }
 }
 
@@ -1293,33 +1319,45 @@ impl<
 
     fn build(
         self,
-    ) -> Result<Session<TransportRustls, RustlsConnectionManager, LB>, SessionBuildError> {
-        verify_compression_configuration(self.node_config.version, self.config.compression).map(
-            |()| {
-                let (keyspace_holder, keyspace_receiver) = create_keyspace_holder();
-                let connection_manager = RustlsConnectionManager::new(
-                    self.node_config.dns_name,
-                    self.node_config.authenticator_provider,
-                    self.node_config.config,
-                    keyspace_holder.clone(),
-                    self.config.reconnection_policy.clone(),
-                    self.frame_encoder_factory,
-                    self.config.compression,
-                    self.config.transport_buffer_size,
-                    self.config.tcp_nodelay,
-                    self.node_config.version,
-                );
+    ) -> BoxFuture<
+        'static,
+        Result<Session<TransportRustls, RustlsConnectionManager, LB>, SessionBuildError>,
+    > {
+        async move {
+            match verify_compression_configuration(
+                self.node_config.version,
+                self.config.compression,
+            ) {
+                Ok(()) => {
+                    let (keyspace_holder, keyspace_receiver) = create_keyspace_holder();
+                    let connection_manager = RustlsConnectionManager::new(
+                        self.node_config.dns_name,
+                        self.node_config.authenticator_provider,
+                        self.node_config.config,
+                        keyspace_holder.clone(),
+                        self.config.reconnection_policy.clone(),
+                        self.frame_encoder_factory,
+                        self.config.compression,
+                        self.config.transport_buffer_size,
+                        self.config.tcp_nodelay,
+                        self.node_config.version,
+                    );
 
-                self.config.into_session(
-                    keyspace_holder,
-                    keyspace_receiver,
-                    self.node_config.contact_points,
-                    connection_manager,
-                    self.node_config.version,
-                    self.node_config.beta_protocol,
-                )
-            },
-        )
+                    self.config
+                        .into_session(
+                            keyspace_holder,
+                            keyspace_receiver,
+                            self.node_config.contact_points,
+                            connection_manager,
+                            self.node_config.version,
+                            self.node_config.beta_protocol,
+                        )
+                        .await
+                }
+                Err(err) => Err(err),
+            }
+        }
+        .boxed()
     }
 }
 
diff --git a/cdrs-tokio/src/lib.rs b/cdrs-tokio/src/lib.rs
index 427f7a4..60a296c 100644
--- a/cdrs-tokio/src/lib.rs
+++ b/cdrs-tokio/src/lib.rs
@@ -19,6 +19,7 @@
 //!         .unwrap();
 //!     let session = TcpSessionBuilder::new(RoundRobinLoadBalancingStrategy::new(), cluster_config)
 //!         .build()
+//!         .await
 //!         .unwrap();
 //!
 //!     let create_ks = "CREATE KEYSPACE IF NOT EXISTS test_ks WITH REPLICATION = { \
diff --git a/cdrs-tokio/src/retry/retry_policy.rs b/cdrs-tokio/src/retry/retry_policy.rs
index cafc0ff..93a7c1c 100644
--- a/cdrs-tokio/src/retry/retry_policy.rs
+++ b/cdrs-tokio/src/retry/retry_policy.rs
@@ -2,7 +2,7 @@ use derive_more::Display;
 
 use cassandra_protocol::error::Error;
 use cassandra_protocol::frame::message_error::{
-    AdditionalErrorInfo, ErrorBody, ReadTimeoutError, WriteTimeoutError, WriteType,
+    ErrorBody, ErrorType, ReadTimeoutError, WriteTimeoutError, WriteType,
 };
 
 #[derive(Debug, PartialEq, Eq, Ord, PartialOrd, Hash, Copy, Clone, Display)]
@@ -75,7 +75,7 @@ impl RetrySession for DefaultRetrySession {
             | Error::Server {
                 body:
                     ErrorBody {
-                        additional_info: AdditionalErrorInfo::Overloaded,
+                        ty: ErrorType::Overloaded,
                         ..
                     },
                 ..
@@ -83,7 +83,7 @@ impl RetrySession for DefaultRetrySession {
             | Error::Server {
                 body:
                     ErrorBody {
-                        additional_info: AdditionalErrorInfo::Server,
+                        ty: ErrorType::Server,
                         ..
                     },
                 ..
@@ -91,7 +91,7 @@ impl RetrySession for DefaultRetrySession {
             | Error::Server {
                 body:
                     ErrorBody {
-                        additional_info: AdditionalErrorInfo::Truncate,
+                        ty: ErrorType::Truncate,
                         ..
                     },
                 ..
@@ -105,7 +105,7 @@ impl RetrySession for DefaultRetrySession {
             Error::Server {
                 body:
                     ErrorBody {
-                        additional_info: AdditionalErrorInfo::Unavailable(_),
+                        ty: ErrorType::Unavailable(_),
                         ..
                     },
                 ..
@@ -120,8 +120,7 @@ impl RetrySession for DefaultRetrySession {
             Error::Server {
                 body:
                     ErrorBody {
-                        additional_info:
-                            AdditionalErrorInfo::ReadTimeout(error @ ReadTimeoutError { .. }),
+                        ty: ErrorType::ReadTimeout(error @ ReadTimeoutError { .. }),
                         ..
                     },
                 ..
@@ -139,8 +138,7 @@ impl RetrySession for DefaultRetrySession {
             Error::Server {
                 body:
                     ErrorBody {
-                        additional_info:
-                            AdditionalErrorInfo::WriteTimeout(error @ WriteTimeoutError { .. }),
+                        ty: ErrorType::WriteTimeout(error @ WriteTimeoutError { .. }),
                         ..
                     },
                 ..
@@ -158,7 +156,7 @@ impl RetrySession for DefaultRetrySession {
             Error::Server {
                 body:
                     ErrorBody {
-                        additional_info: AdditionalErrorInfo::IsBootstrapping,
+                        ty: ErrorType::IsBootstrapping,
                         ..
                     },
                 ..


Test patch:
diff --git a/cdrs-tokio/tests/common.rs b/cdrs-tokio/tests/common.rs
index 28ea98f..94e8e46 100644
--- a/cdrs-tokio/tests/common.rs
+++ b/cdrs-tokio/tests/common.rs
@@ -52,6 +52,7 @@ pub async fn setup_multiple(
     let session = TcpSessionBuilder::new(RoundRobinLoadBalancingStrategy::new(), cluster_config)
         .with_reconnection_policy(Arc::new(NeverReconnectionPolicy::default()))
         .build()
+        .await
         .unwrap();
     let re_table_name = Regex::new(r"CREATE TABLE IF NOT EXISTS (\w+\.\w+)").unwrap();
 
diff --git a/cdrs-tokio/tests/compression.rs b/cdrs-tokio/tests/compression.rs
index 1b0600f..c71775a 100644
--- a/cdrs-tokio/tests/compression.rs
+++ b/cdrs-tokio/tests/compression.rs
@@ -38,6 +38,7 @@ async fn encode_decode_test(version: Version) {
         .with_reconnection_policy(Arc::new(NeverReconnectionPolicy::default()))
         .with_compression(Compression::Lz4)
         .build()
+        .await
         .unwrap();
 
     session
diff --git a/cdrs-tokio/tests/keyspace.rs b/cdrs-tokio/tests/keyspace.rs
index 6659810..f06d476 100644
--- a/cdrs-tokio/tests/keyspace.rs
+++ b/cdrs-tokio/tests/keyspace.rs
@@ -34,6 +34,7 @@ async fn create_keyspace() {
     let session = TcpSessionBuilder::new(lb, cluster_config)
         .with_reconnection_policy(Arc::new(NeverReconnectionPolicy::default()))
         .build()
+        .await
         .unwrap();
 
     let drop_query = "DROP KEYSPACE IF EXISTS create_ks_test";
@@ -107,6 +108,7 @@ async fn alter_keyspace() {
     let session = TcpSessionBuilder::new(lb, cluster_config)
         .with_reconnection_policy(Arc::new(NeverReconnectionPolicy::default()))
         .build()
+        .await
         .unwrap();
 
     let drop_query = "DROP KEYSPACE IF EXISTS alter_ks_test";
@@ -171,6 +173,7 @@ async fn use_keyspace() {
     let session = TcpSessionBuilder::new(lb, cluster_config)
         .with_reconnection_policy(Arc::new(NeverReconnectionPolicy::default()))
         .build()
+        .await
         .unwrap();
 
     let create_query = "CREATE KEYSPACE IF NOT EXISTS use_ks_test WITH \
@@ -208,6 +211,7 @@ async fn drop_keyspace() {
     let session = TcpSessionBuilder::new(lb, cluster_config)
         .with_reconnection_policy(Arc::new(NeverReconnectionPolicy::default()))
         .build()
+        .await
         .unwrap();
 
     let create_query = "CREATE KEYSPACE IF NOT EXISTS drop_ks_test WITH \
diff --git a/cdrs-tokio/tests/multi_node_speculative_execution.rs b/cdrs-tokio/tests/multi_node_speculative_execution.rs
index 5f78390..3035085 100644
--- a/cdrs-tokio/tests/multi_node_speculative_execution.rs
+++ b/cdrs-tokio/tests/multi_node_speculative_execution.rs
@@ -40,6 +40,7 @@ async fn multi_node_speculative_execution() {
             Duration::from_secs(0),
         )))
         .build()
+        .await
         .unwrap();
 
     let create_keyspace_query = "CREATE KEYSPACE IF NOT EXISTS cdrs_test WITH \
diff --git a/cdrs-tokio/tests/multithread.rs b/cdrs-tokio/tests/multithread.rs
index a0cb9a3..7e963c0 100644
--- a/cdrs-tokio/tests/multithread.rs
+++ b/cdrs-tokio/tests/multithread.rs
@@ -24,6 +24,7 @@ async fn multithread() {
         TcpSessionBuilder::new(RoundRobinLoadBalancingStrategy::new(), cluster_config)
             .with_reconnection_policy(Arc::new(NeverReconnectionPolicy::default()))
             .build()
+            .await
             .unwrap();
 
     no_compression.query("CREATE KEYSPACE IF NOT EXISTS test_ks WITH REPLICATION = { 'class' : 'SimpleStrategy', 'replication_factor' : 1 };").await.expect("Could not create ks");
diff --git a/cdrs-tokio/tests/paged_query.rs b/cdrs-tokio/tests/paged_query.rs
index f9bfd93..176cff7 100644
--- a/cdrs-tokio/tests/paged_query.rs
+++ b/cdrs-tokio/tests/paged_query.rs
@@ -24,6 +24,7 @@ async fn paged_query() {
     let session = TcpSessionBuilder::new(lb, cluster_config)
         .with_reconnection_policy(Arc::new(NeverReconnectionPolicy::default()))
         .build()
+        .await
         .unwrap();
 
     session
diff --git a/cdrs-tokio/tests/single_node_speculative_execution.rs b/cdrs-tokio/tests/single_node_speculative_execution.rs
index a870186..f2a93eb 100644
--- a/cdrs-tokio/tests/single_node_speculative_execution.rs
+++ b/cdrs-tokio/tests/single_node_speculative_execution.rs
@@ -38,6 +38,7 @@ async fn single_node_speculative_execution() {
             Duration::from_secs(1),
         )))
         .build()
+        .await
         .unwrap();
 
     let create_keyspace_query = "CREATE KEYSPACE IF NOT EXISTS cdrs_test WITH \
diff --git a/cdrs-tokio/tests/topology_aware.rs b/cdrs-tokio/tests/topology_aware.rs
index 5c68e8d..622a3d3 100644
--- a/cdrs-tokio/tests/topology_aware.rs
+++ b/cdrs-tokio/tests/topology_aware.rs
@@ -43,6 +43,7 @@ async fn query_topology_aware() {
         "datacenter1".into(),
     )))
     .build()
+    .await
     .unwrap();
 
     let create_keyspace_query = "CREATE KEYSPACE IF NOT EXISTS cdrs_test WITH \