Repository: AlexPikalov/cdrs (language: rust)

Issue:
Protocol parsing issues on many inserts
Given the schema
```sql
CREATE KEYSPACE meow WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1};
USE meow;
CREATE TABLE kittehz (id UUID PRIMARY KEY, name VARCHAR, stuff BOOLEAN);
```

running the latest Cassandra version from Docker on macOS HS
```
docker run --name some-cassandra -d -p 9042:9042 cassandra
```

```
cqlsh:meow> SHOW VERSION;
[cqlsh 5.0.1 | Cassandra 3.11.2 | CQL spec 3.4.4 | Native protocol v4]
```

executing the code
```rust
extern crate cdrs;

use cdrs::client::CDRS;
use cdrs::authenticators::NoneAuthenticator;
use cdrs::transport::TransportTcp;
use cdrs::query::QueryBuilder;
use cdrs::compression::Compression;
use std::error::Error;


fn main() -> Result<(), Box<Error>> {
    let authenticator = NoneAuthenticator;
    let addr = "127.0.0.1:9042";
    let tcp_transport = TransportTcp::new(addr).unwrap();

    // pass authenticator and transport into CDRS' constructor
    let client = CDRS::new(tcp_transport, authenticator);

    // start session without compression
    let session = client.start(Compression::None)?;

    match session.query(QueryBuilder::new("USE meow").finalize(), false, false) {
        Err(ref err) => panic!("cant use keyspace {:?}", err),
        Ok(_) => {},
    }

    for i in 0..100_000 {
        let query = QueryBuilder::new(r#"
            INSERT INTO kittehz
            (id, name, stuff)
            VALUES
            (uuid(), 'a', false)
        "#).finalize();

        match session.query(query, false, false) {
            Ok(_) => {},
            Err(ref err) => panic!("cant insert {:?}", err),
        }
    }

    Ok(())
}
```

crashes the library after 15123 inserts, either entering unreachable code

```
➜  cdrs-bug git:(master) ✗ cargo run --release
   Compiling cdrs-bug v0.1.0 (file:///Users/ath/Desktop/cdrs-bug)
    Finished release [optimized] target(s) in 0.44s
     Running `target/release/cdrs-bug`
thread 'main' panicked at 'internal error: entered unreachable code', /Users/ath/.cargo/registry/src/github.com-1ecc6299db9ec823/cdrs-1.2.3/src/frame/mod.rs:322:18
note: Run with `RUST_BACKTRACE=1` for a backtrace.
```

or reporting an unexpected Cassandra version
```

➜  cdrs-bug git:(master) ✗ cargo run --release
    Finished release [optimized] target(s) in 0.02s
     Running `target/release/cdrs-bug`
thread 'main' panicked at 'Unexpected Cassandra version 0, either 4 or 132 is expected', /Users/ath/.cargo/registry/src/github.com-1ecc6299db9ec823/cdrs-1.2.3/src/frame/mod.rs:166:13
note: Run with `RUST_BACKTRACE=1` for a backtrace.
```

Looks like the parser is getting confused, reading from a wrong offset. I attached a Cargo module / project (what do you even call them?) that reproduces the issue on my machine™.

[cdrs-bug.tar.gz](https://github.com/AlexPikalov/cdrs/files/2006420/cdrs-bug.tar.gz)


Discussion / hints:
The issue also persists with `v2.0.0`, with a slightly different error:
```
➜  cdrs-bug cargo run --release
    Finished release [optimized] target(s) in 0.03s
     Running `target/release/cdrs-bug`
thread 'main' panicked at 'cant insert General("Empty or corrupted frame received")', src/main.rs:38:29
note: Run with `RUST_BACKTRACE=1` for a backtrace.
```

```rust
extern crate cdrs;

use std::collections::HashMap;

use cdrs::authenticators::NoneAuthenticator;
use cdrs::cluster::{Cluster, Session};
use cdrs::query::*;
use cdrs::load_balancing::{RoundRobin};
use cdrs::transport::TransportTcp;

use cdrs::types::prelude::*;
use cdrs::types::from_cdrs::FromCDRSByName;
use cdrs::frame::IntoBytes;
use std::error::Error;


fn main() -> Result<(), Box<Error>> {
    let cluster = Cluster::new(vec!["127.0.0.1:9042"], NoneAuthenticator {});
    let mut session = cluster
        .connect(RoundRobin::new())
        .expect("No compression connection error");

    match session.query("USE meow") {
        Err(ref err) => panic!("cant use keyspace {:?}", err),
        Ok(_) => {},
    }

    for _ in 0..100_000 {
        match session.query(r#"
            INSERT INTO kittehz
            (id, name, stuff)
            VALUES
            (uuid(), 'a', false)
        "#) {
            Ok(_) => {},
            Err(ref err) => panic!("cant insert {:?}", err),
        }
    }

    Ok(())
}
```


@athre0z Thank you for reporting this issue. I'm investigating it currently.
@athre0z Unfortunately I was not able to reproduce it so far after 5 times running the code.
Last time the fix for very similar issue was to substitute `read` by `read_exact` in [parser.rs](https://github.com/AlexPikalov/cdrs/blob/master/src/frame/parser.rs).

Would you be able to check possible solution by replacing following code in `parser.rs`

```rust
    // NOTE: order of reads matters
    let v = try!(cursor.read(&mut version_bytes));
    let f = try!(cursor.read(&mut flag_bytes));
    let s = try!(cursor.read(&mut stream_bytes));
    let o = try!(cursor.read(&mut opcode_bytes));
    let l = try!(cursor.read(&mut length_bytes));

    if v != Version::BYTE_LENGTH || f != Flag::BYTE_LENGTH || s != STREAM_LEN
       || o != Opcode::BYTE_LENGTH || l != LENGTH_LEN {
        return Err(error::Error::from("Empty or corrupted frame received"));
    }
```

by the possible fix

```rust
    try!(cursor.read_exact(&mut version_bytes));
    try!(cursor.read_exact(&mut flag_bytes));
    try!(cursor.read_exact(&mut stream_bytes));
    try!(cursor.read_exact(&mut opcode_bytes));
    try!(cursor.read_exact(&mut length_bytes));
```

and use your local version of CDRS (cargo.toml):

```toml
[dependencies.cdrs]
path = "../path-to-cdrs"
```

Please let me know if it helped. If so please submit a PR with this solution.

Thank you.
Thank you for the fast response, that indeed fixes the issue! I'll wrap it into a PR shortly. A quick grep through the code revealed another use of the plain `read()` in types/mod.rs `CStringList::from_cursor`. Should I substitute this one as well? It does look like we'd always want to read 2 bytes here as well.

Fix patch:
diff --git a/src/frame/parser.rs b/src/frame/parser.rs
index 1bac77f..e2dfcc6 100644
--- a/src/frame/parser.rs
+++ b/src/frame/parser.rs
@@ -16,16 +16,11 @@ pub fn parse_frame(cursor: &mut Read, compressor: &Compression) -> error::Result
     let mut length_bytes = [0; LENGTH_LEN];
 
     // NOTE: order of reads matters
-    let v = try!(cursor.read(&mut version_bytes));
-    let f = try!(cursor.read(&mut flag_bytes));
-    let s = try!(cursor.read(&mut stream_bytes));
-    let o = try!(cursor.read(&mut opcode_bytes));
-    let l = try!(cursor.read(&mut length_bytes));
-
-    if v != Version::BYTE_LENGTH || f != Flag::BYTE_LENGTH || s != STREAM_LEN
-       || o != Opcode::BYTE_LENGTH || l != LENGTH_LEN {
-        return Err(error::Error::from("Empty or corrupted frame received"));
-    }
+    try!(cursor.read_exact(&mut version_bytes));
+    try!(cursor.read_exact(&mut flag_bytes));
+    try!(cursor.read_exact(&mut stream_bytes));
+    try!(cursor.read_exact(&mut opcode_bytes));
+    try!(cursor.read_exact(&mut length_bytes));
 
     let version = Version::from(version_bytes.to_vec());
     let flags = Flag::get_collection(flag_bytes[0]);
diff --git a/src/types/mod.rs b/src/types/mod.rs
index 3d0ab7a..1a1b7ce 100644
--- a/src/types/mod.rs
+++ b/src/types/mod.rs
@@ -464,7 +464,7 @@ impl FromCursor for CStringList {
     fn from_cursor(mut cursor: &mut Cursor<&[u8]>) -> CDRSResult<CStringList> {
         // TODO: try to use slice instead
         let mut len_bytes = [0; SHORT_LEN];
-        try!(cursor.read(&mut len_bytes));
+        try!(cursor.read_exact(&mut len_bytes));
         let len = try_from_bytes(len_bytes.to_vec().as_slice())? as usize;
         let mut list = Vec::with_capacity(len);
         for _ in 0..len {


Test patch: