Repository: AlexPikalov/cdrs (language: rust)

Issue:
No prepared statement with ID {} found.
Hi!
Get this error when trying to update a CF multiple times in Scylla.
I filed an issued to scylla and attached a tcpdump showing the traffic between the driver and the database.
The Scylla team said it's a problem with the driver that's ignoring the error coming from Scylla (seems there's a cache for prepared statements and from time to time it gets evicted) instead of retrying to reprepare the statement.


Discussion / hints:
Hi @bruno-barin-ifood ,
Can you please provide me with a code snipped that reproduces this behaviour?
Sure, here you are:

```use std::str;

use cdrs::frame::Frame;
use cdrs::query::*;
use cdrs::types::prelude::*;
use log::error;
use rdkafka::consumer::Consumer;
use rdkafka::Message;

use model::Cart;
use settings::*;

mod model;
mod settings;

fn main() {
    env_logger::init();
    consume_messages();
}

const UPDATE_QUERY: &str = "UPDATE cart.carts SET cart = ?, status = ? WHERE id = ?";

struct Scylla {
    prepared_query: PreparedQuery,
    scylla_session: ScyllaSession
}

impl Scylla {
    fn new() -> Self {
        let prepared_query = self.scylla_session.prepare(UPDATE_QUERY).unwrap();
        let scylla_session = get_scylla_session();
        Scylla {
            prepared_query,
            scylla_session
        }
    }
    fn update_scylla(&self, cart: &Cart) -> Result<Frame> {

        self.scylla_session.exec_with_values(&self.prepared_query, cart.into_query_values())
    }
}

fn consume_messages() {
    let kafka_consumer = kafka_consumer();
    let scylla = Scylla::new();

    for potential_message in kafka_consumer.iter() {
        match potential_message {
            Err(e) => error!("Error to fetch message from Kafka. Reason: {}", e),
            Ok(message) => {
                let body = message.payload_view().unwrap().unwrap();
                let cart: Cart = serde_json::from_str(body).unwrap();
                let result = scylla.update_scylla(&cart);
                match result {
                    Ok(_) => {
                        kafka_consumer.store_offset(&message)
                            .expect("Failed to commit message to Kafka");
                    }
                    Err(_err) => println!("Cart {} failed due to connection problems - {}", cart.id, _err)
                }
            }
        }
    }
}
```
The link for the issue I've opened to the Scylla team is this one:

https://github.com/scylladb/scylla/issues/5936

@bruno-barin-ifood 
If you have a cluster with more than one Scylla/Cassandra nodes it may be problematic to execute prepared queries with current CDRS implementation - during the next request to cluster next or random node (depending on the balancing strategy) will be used so it may lead to the situation when such not doesn't have a corresponded prepared query registered.
So it that regards CDRS may need some improvements
thanks @AlexPikalov !
Please note this can happen also in a single node, that is a Cassandra/Scylla node may invalidate the prepared statements (due to some internal reasons).

When a prepared statement will be executed  the error of not having the prepared statement will be returned (https://github.com/apache/cassandra/blob/trunk/doc/native_protocol_v4.spec#L1169)

Most drivers at this point reprepare the statement and following that reexecute it.

So while it would be preferable to prepare a statement on all connections (to make sure its avail) yet this is not a must, it would be sufficient to handle the error and reexecute the statement.


@bruno-barin-ifood I've just read the conversation inside the issue that you've created for Scylla.

@slivne Got you. It's an interesting fact which was not known to me before.

I think I need to think how to make the driver smarter in that regards.
Hi @AlexPikalov do you have any progress with this issue?

is it possible to add a specific error for this?
I guess a work around for now is to just query directly with the prepared statement.

Fix patch:
diff --git a/src/frame/frame_batch.rs b/src/frame/frame_batch.rs
index a2656a44..1cef5091 100644
--- a/src/frame/frame_batch.rs
+++ b/src/frame/frame_batch.rs
@@ -2,7 +2,7 @@ use rand;
 
 use crate::consistency::Consistency;
 use crate::frame::*;
-use crate::query::QueryFlags;
+use crate::query::{QueryFlags, PreparedQuery};
 use crate::query::QueryValues;
 use crate::types::*;
 
@@ -107,7 +107,7 @@ pub struct BatchQuery {
 /// It contains either an id of prepared query or CQL string.
 #[derive(Debug, Clone)]
 pub enum BatchQuerySubj {
-    PreparedId(CBytesShort),
+    PreparedId(PreparedQuery),
     QueryString(CStringLong),
 }
 
@@ -124,7 +124,7 @@ impl IntoBytes for BatchQuery {
 
         match self.subject {
             BatchQuerySubj::PreparedId(ref s) => {
-                bytes.extend_from_slice(s.into_cbytes().as_slice());
+                bytes.extend_from_slice(s.id.borrow().into_cbytes().as_slice());
             }
             BatchQuerySubj::QueryString(ref s) => {
                 bytes.extend_from_slice(s.into_cbytes().as_slice());
diff --git a/src/frame/frame_execute.rs b/src/frame/frame_execute.rs
index c54e40b7..8f0e26fd 100644
--- a/src/frame/frame_execute.rs
+++ b/src/frame/frame_execute.rs
@@ -11,12 +11,12 @@ pub struct BodyReqExecute<'a> {
     id: &'a CBytesShort,
     /// Query paramaters which have the same meaning as one for `query`
     /// TODO: clarify if it is QueryParams or its shortened variant
-    query_parameters: QueryParams,
+    query_parameters: &'a QueryParams,
 }
 
 impl<'a> BodyReqExecute<'a> {
     /// The method which creates new instance of `BodyReqExecute`
-    pub fn new(id: &CBytesShort, query_parameters: QueryParams) -> BodyReqExecute {
+    pub fn new<'b>(id: &'b CBytesShort, query_parameters: &'b QueryParams) -> BodyReqExecute<'b> {
         BodyReqExecute {
             id: id,
             query_parameters: query_parameters,
@@ -37,7 +37,7 @@ impl Frame {
     /// **Note:** This function should be used internally for building query request frames.
     pub fn new_req_execute(
         id: &CBytesShort,
-        query_parameters: QueryParams,
+        query_parameters: &QueryParams,
         flags: Vec<Flag>,
     ) -> Frame {
         let version = Version::Request;
diff --git a/src/query/batch_query_builder.rs b/src/query/batch_query_builder.rs
index b0db2949..3d0f4226 100644
--- a/src/query/batch_query_builder.rs
+++ b/src/query/batch_query_builder.rs
@@ -1,8 +1,8 @@
 use crate::consistency::Consistency;
 use crate::error::{Error as CError, Result as CResult};
 use crate::frame::frame_batch::{BatchQuery, BatchQuerySubj, BatchType, BodyReqBatch};
-use crate::query::{QueryFlags, QueryValues};
-use crate::types::{CBytesShort, CStringLong};
+use crate::query::{QueryFlags, QueryValues, PreparedQuery};
+use crate::types::CStringLong;
 
 pub type QueryBatch = BodyReqBatch;
 
@@ -42,10 +42,10 @@ impl BatchQueryBuilder {
     }
 
     /// Add a query (prepared one)
-    pub fn add_query_prepared(mut self, query_id: CBytesShort, values: QueryValues) -> Self {
+    pub fn add_query_prepared(mut self, query: PreparedQuery, values: QueryValues) -> Self {
         self.queries.push(BatchQuery {
             is_prepared: true,
-            subject: BatchQuerySubj::PreparedId(query_id),
+            subject: BatchQuerySubj::PreparedId(query),
             values,
         });
         self
diff --git a/src/query/exec_executor.rs b/src/query/exec_executor.rs
index f6dba935..fd96f7c3 100644
--- a/src/query/exec_executor.rs
+++ b/src/query/exec_executor.rs
@@ -4,18 +4,16 @@ use std::cell::RefCell;
 use crate::cluster::{GetCompressor, GetConnection};
 use crate::error;
 use crate::frame::{Frame, IntoBytes};
-use crate::query::{QueryParams, QueryParamsBuilder, QueryValues};
+use crate::query::{QueryParams, QueryParamsBuilder, QueryValues, PreparedQuery, PrepareExecutor};
 use crate::transport::CDRSTransport;
-use crate::types::CBytesShort;
 
 use super::utils::{prepare_flags, send_frame};
-
-pub type PreparedQuery = CBytesShort;
+use crate::error::Error;
 
 pub trait ExecExecutor<
     T: CDRSTransport + 'static,
     M: r2d2::ManageConnection<Connection = RefCell<T>, Error = error::Error> + Sized,
->: GetConnection<T, M> + GetCompressor<'static>
+>: GetConnection<T, M> + GetCompressor<'static> + PrepareExecutor<T, M>
 {
     fn exec_with_params_tw(
         &self,
@@ -28,9 +26,21 @@ pub trait ExecExecutor<
         Self: Sized,
     {
         let flags = prepare_flags(with_tracing, with_warnings);
-        let options_frame = Frame::new_req_execute(prepared, query_parameters, flags).into_cbytes();
+        let options_frame = Frame::new_req_execute(&prepared.id.borrow(), &query_parameters, flags).into_cbytes();
 
-        send_frame(self, options_frame)
+        let mut result = send_frame(self, options_frame);
+        if let Err(Error::Server(error)) = &result {
+            // if query is unprepared
+            if error.error_code == 0x2500 {
+                if let Ok(new) = self.prepare_raw(&prepared.query) {
+                    prepared.id.replace(new.id);
+                    let flags = prepare_flags(with_tracing, with_warnings);
+                    let options_frame = Frame::new_req_execute(&prepared.id.borrow(), &query_parameters, flags).into_cbytes();
+                    result = send_frame(self, options_frame);
+                }
+            }
+        }
+        result
     }
 
     fn exec_with_params(
diff --git a/src/query/mod.rs b/src/query/mod.rs
index 18684ff0..fd863c0b 100644
--- a/src/query/mod.rs
+++ b/src/query/mod.rs
@@ -2,6 +2,7 @@ mod batch_executor;
 mod batch_query_builder;
 mod exec_executor;
 mod prepare_executor;
+mod prepared_query;
 mod query;
 mod query_executor;
 mod query_flags;
@@ -13,7 +14,8 @@ mod utils;
 pub use crate::query::batch_executor::BatchExecutor;
 pub use crate::query::batch_query_builder::{BatchQueryBuilder, QueryBatch};
 pub use crate::query::exec_executor::ExecExecutor;
-pub use crate::query::prepare_executor::{PrepareExecutor, PreparedQuery};
+pub use crate::query::prepare_executor::PrepareExecutor;
+pub use crate::query::prepared_query::PreparedQuery;
 pub use crate::query::query::Query;
 pub use crate::query::query_executor::QueryExecutor;
 pub use crate::query::query_flags::QueryFlags;
diff --git a/src/query/prepare_executor.rs b/src/query/prepare_executor.rs
index c121b458..e6b7e72b 100644
--- a/src/query/prepare_executor.rs
+++ b/src/query/prepare_executor.rs
@@ -6,11 +6,9 @@ use crate::error;
 use crate::frame::frame_result::BodyResResultPrepared;
 use crate::frame::{Frame, IntoBytes};
 use crate::transport::CDRSTransport;
-use crate::types::CBytesShort;
 
 use super::utils::{prepare_flags, send_frame};
-
-pub type PreparedQuery = CBytesShort;
+use crate::query::PreparedQuery;
 
 pub trait PrepareExecutor<
     T: CDRSTransport + 'static,
@@ -65,8 +63,9 @@ pub trait PrepareExecutor<
     where
         Self: Sized,
     {
+        let str = query.to_string();
         self.prepare_raw_tw(query, with_tracing, with_warnings)
-            .map(|x| x.id)
+            .map(|x| PreparedQuery { id: RefCell::new(x.id), query: str })
     }
 
     /// It prepares query without additional tracing information and warnings.
diff --git a/src/query/prepared_query.rs b/src/query/prepared_query.rs
new file mode 100644
index 00000000..016b881c
--- /dev/null
+++ b/src/query/prepared_query.rs
@@ -0,0 +1,8 @@
+use std::cell::RefCell;
+use crate::types::CBytesShort;
+
+#[derive(Debug, Clone)]
+pub struct PreparedQuery {
+	pub(crate) id: RefCell<CBytesShort>,
+	pub(crate) query: String,
+}
diff --git a/src/query/query_params.rs b/src/query/query_params.rs
index eabd9c5f..452cf417 100644
--- a/src/query/query_params.rs
+++ b/src/query/query_params.rs
@@ -6,7 +6,7 @@ use crate::query::query_values::QueryValues;
 use crate::types::{to_bigint, to_int, to_short, CBytes};
 
 /// Parameters of Query for query operation.
-#[derive(Debug, Default)]
+#[derive(Debug, Default, Clone)]
 pub struct QueryParams {
     /// Cassandra consistency level.
     pub consistency: Consistency,


Test patch: