StreamReadReply silently discards data

[mxbrt]

I am trying to write and read a redis stream using xadd and xread. While I can see that the data exists in redis, I get an empty StreamReadReply from xread_options:

let _: () = redis::pipe()
    .xadd("stream", "*", &[("key", "val")])
    .query_async(&mut conn)
    .await?;
let options = StreamReadOptions::default().block(0).count(1);
let reply: StreamReadReply = redis::pipe()
    .xread_options(&["stream"], &["0"], &options)
    .query_async(&mut conn)
    .await?;
eprintln!("reply {:?}", reply); // Prints: reply StreamReadReply { keys: [] }

If I do not convert the query into a StreamReadReply, but print it as a redis::Value, i can see that some data is returned, but it is wrapped in one more Value::Bulk then redis-rs expects. The data is silently discarded during conversion to StreamReadReply.

Example:

use anyhow::Result;
use redis::{streams::StreamReadReply, FromRedisValue, Value};

fn main() -> Result<()> {
    // This is the data that I get from redis
    // bulk(bulk(bulk(string-data('"stream"'), bulk(bulk(string-data('"0-1"'), bulk(string-data('"key"'), string-data('val'))))))))
    let v1 = Value::Bulk(vec![Value::Bulk(vec![Value::Bulk(vec![
        Value::Data("stream".into()),
        Value::Bulk(vec![Value::Bulk(vec![
            Value::Data("0-1".into()),
            Value::Bulk(vec![Value::Data("key".into()), Value::Data("val".into())]),
        ])]),
    ])])]);
    // returns an empty StreamReadReply
    let v1 = StreamReadReply::from_redis_value(&v1)?;
    // This is the data how redis-rs expects it
    // bulk(bulk(string-data('"stream"'), bulk(bulk(string-data('"0-1"'), bulk(string-data('"key"'), string-data('val')))))))
    let v2 = Value::Bulk(vec![Value::Bulk(vec![
        Value::Data("stream".into()),
        Value::Bulk(vec![Value::Bulk(vec![
            Value::Data("0-1".into()),
            Value::Bulk(vec![Value::Data("key".into()), Value::Data("val".into())]),
        ])]),
    ])]);
    let v2 = StreamReadReply::from_redis_value(&v2)?;
    eprintln!("\nv1 {:?}\nv2 {:?}\n", v1, v2);

    // output:
    // v1 StreamReadReply { keys: [] }
    // v2 StreamReadReply { keys: [StreamKey { key: "stream", ids: [StreamId { id: "0-1", map: {"key": string-data('"val"')} }] }] }
    Ok(())
}

I would expect redis-rs to at least return an error instead of discarding the data.

[jaymell]

I think the issue is you're using pipe. If you issue the command directly against the connection, I don't think you'll see this issue.

[mxbrt]

Thanks, that seems to be the issue. But is it expected behavior that an empty reply is parsed from the response data, instead of an error?

[jaymell]

If you define your return type as Vec<StreamReadReply>, you should also get something more like what you would expect.

But is it expected behavior that an empty reply is parsed from the response data, instead of an error?

It makes sense to me that a value of an unexpected dimension would trigger an error in the FromRedisValue implementation of StreamReadReply. Any interest in submitting a PR?

[mxbrt]

Sure . Root cause seems to be in HashMap conversion #978.