rustis 0.27.2

Redis async driver for Rust
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
use crate::{
    ClientError, Error, ErrorKind, RedisError, RedisErrorKind, Result, TimeoutKind,
    commands::StringCommands,
};
use bytes::Bytes;
use serial_test::serial;

fn redis(kind: RedisErrorKind) -> Error {
    Error::from(ErrorKind::Redis(RedisError {
        kind,
        description: Bytes::new(),
    }))
}

fn client(client_error: ClientError) -> Error {
    Error::from(ErrorKind::Client(client_error))
}

fn timeout(kind: TimeoutKind) -> Error {
    Error::from(ErrorKind::Timeout(kind))
}

fn io() -> Error {
    Error::from(ErrorKind::IO(std::sync::Arc::new(std::io::Error::new(
        std::io::ErrorKind::ConnectionReset,
        "reset",
    ))))
}

#[test]
fn moved_error() {
    let raw_error = b"MOVED 3999 127.0.0.1:6381";
    let error = RedisError::try_from(&raw_error[..]);
    println!("error: {error:?}");
    assert!(matches!(
        error,
        Ok(RedisError {
            kind: RedisErrorKind::Moved { hash_slot: 3999, address: (host, 6381) },
            description
        }) if description.is_empty() && host == "127.0.0.1"
    ));
}

#[test]
fn ask_error() {
    let raw_error = b"ASK 3999 127.0.0.1:6381";
    let error = RedisError::try_from(&raw_error[..]);
    assert!(matches!(
        error,
        Ok(RedisError {
            kind: RedisErrorKind::Ask { hash_slot: 3999, address: (host, 6381) },
            description
        }) if description.is_empty() && host == "127.0.0.1"
    ));
}

#[test]
fn moved_error_ipv6() {
    // The address must be split at the last colon (the port separator), so
    // that IPv6 hosts, which contain colons, are parsed correctly.
    let raw_error = b"MOVED 3999 2001:db8::1:6380";
    let error = RedisError::try_from(&raw_error[..]);
    println!("error: {error:?}");
    assert!(matches!(
        error,
        Ok(RedisError {
            kind: RedisErrorKind::Moved { hash_slot: 3999, address: (host, 6380) },
            description
        }) if description.is_empty() && host == "2001:db8::1"
    ));
}

#[test]
fn an_error_carries_no_command_until_one_is_attached() {
    let error = Error::from(ErrorKind::Timeout(TimeoutKind::Command));

    assert!(matches!(error.kind(), ErrorKind::Timeout(_)));
    assert_eq!(None, error.command());
    assert!(error.context().is_none());
    assert_eq!(
        ErrorKind::Timeout(TimeoutKind::Command).to_string(),
        error.to_string()
    );
}

#[test]
fn attaching_a_command_names_it_in_the_context_and_the_message() {
    let error = Error::from(ErrorKind::Timeout(TimeoutKind::Command))
        .with_command(Bytes::from_static(b"BLMPOP"));

    assert_eq!(Some("BLMPOP"), error.command());
    assert_eq!("BLMPOP", error.context().unwrap().command());
    assert!(
        error.to_string().contains("BLMPOP"),
        "the rendered message must name the command, got {error}"
    );
    // The variant stays reachable: attaching context is not a variant change.
    assert!(matches!(error.kind(), ErrorKind::Timeout(_)));
}

/// The site closest to the cause holds the best command, so an outer layer
/// never overwrites what an inner one already attached.
#[test]
fn the_innermost_command_wins() {
    let error = Error::from(ErrorKind::Timeout(TimeoutKind::Command))
        .with_command(Bytes::from_static(b"GET"))
        .with_command(Bytes::from_static(b"SET"));

    assert_eq!(Some("GET"), error.command());
}

/// `Error` is what every fallible call in the crate returns, so it has to keep
/// slotting into the ecosystem that consumes errors: `?` into a `Box<dyn Error>`
/// or an `anyhow::Error`, and crossing a task boundary.
#[test]
fn the_error_type_keeps_its_bounds() {
    const fn assert_bounds<T: std::error::Error + Send + Sync + Clone + 'static>() {}
    assert_bounds::<Error>();

    let boxed: Box<dyn std::error::Error + Send + Sync> = Box::new(
        Error::from(ErrorKind::Timeout(TimeoutKind::Command))
            .with_command(Bytes::from_static(b"GET")),
    );
    assert!(boxed.to_string().contains("GET"));
}

/// The connection is what died, so the command never got an answer and the
/// client will have to reconnect. A reply the parser could not decode belongs
/// here too: the byte stream is desynchronized, so the connection is done.
#[test]
fn a_connection_error_is_told_from_a_command_error() {
    assert!(io().is_connection_error());
    assert!(Error::from(ErrorKind::EOF).is_connection_error());
    assert!(Error::from(ErrorKind::DisconnectedByPeer).is_connection_error());
    assert!(client(ClientError::CannotParseInteger).is_connection_error());
    assert!(client(ClientError::UnknownRespTag('@')).is_connection_error());

    // The server answered, and answered an error: the connection is fine.
    assert!(!redis(RedisErrorKind::WrongType).is_connection_error());
    // A decode error raised past framing fails one command, not the stream.
    assert!(!client(ClientError::MismatchedKeySlots).is_connection_error());
    assert!(!client(ClientError::CannotParseBytes).is_connection_error());
    assert!(!timeout(TimeoutKind::Command).is_connection_error());
    assert!(!Error::from(ErrorKind::Aborted).is_connection_error());
}

/// A timeout is its own answer: the command may or may not have run, which is
/// neither a connection failure nor a server refusal.
#[test]
fn a_timeout_is_its_own_class() {
    assert!(timeout(TimeoutKind::Command).is_timeout());
    assert!(timeout(TimeoutKind::Connect).is_timeout());

    assert!(!io().is_timeout());
    assert!(!redis(RedisErrorKind::TryAgain).is_timeout());
    assert!(!timeout(TimeoutKind::Command).is_server_error());
    assert!(!timeout(TimeoutKind::Command).is_connection_error());
}

/// The two deadlines demand opposite answers — try another node, or fail this
/// request — so the error has to say which one expired. One variant with one
/// `Display` string left the caller guessing.
#[test]
fn a_timeout_names_the_deadline_that_expired() {
    assert!(matches!(
        timeout(TimeoutKind::Connect).kind(),
        ErrorKind::Timeout(TimeoutKind::Connect)
    ));
    assert!(matches!(
        timeout(TimeoutKind::Command).kind(),
        ErrorKind::Timeout(TimeoutKind::Command)
    ));

    assert_ne!(
        timeout(TimeoutKind::Connect).to_string(),
        timeout(TimeoutKind::Command).to_string()
    );
}

/// A deadline that expires while the connection is being made is a
/// `connect_timeout`, and one that expires while a command is in flight is a
/// `command_timeout`. Both are driven here against a server that accepts the
/// socket and answers nothing.
#[cfg(feature = "tokio-runtime")]
#[tokio::test]
#[serial]
async fn each_deadline_reports_its_own_kind() -> Result<()> {
    use crate::{
        client::{Client, Config, IntoConfig},
        tests::fake_server::HELLO_REPLY,
    };
    use std::time::Duration;
    use tokio::io::{AsyncReadExt, AsyncWriteExt};

    // `answer_handshake` false: the socket is accepted and nothing is ever sent,
    // so the connection itself cannot complete. True: the handshake is answered
    // and the first command is left waiting.
    async fn silent_server(
        answer_handshake: bool,
    ) -> Result<(std::net::SocketAddr, tokio::task::JoinHandle<()>)> {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
        let addr = listener.local_addr()?;
        let server = tokio::spawn(async move {
            let Ok((mut stream, _)) = listener.accept().await else {
                return;
            };
            let mut chunk = [0u8; 1024];
            if answer_handshake {
                if stream.read(&mut chunk).await.is_err() {
                    return;
                }
                if stream.write_all(HELLO_REPLY).await.is_err() {
                    return;
                }
            }
            while stream.read(&mut chunk).await.is_ok_and(|n| n > 0) {}
        });
        Ok((addr, server))
    }

    let (addr, server) = silent_server(false).await?;
    let mut config: Config = format!("redis://{addr}").into_config()?;
    config.connect_timeout = Duration::from_millis(200);
    let Err(error) = Client::connect(config).await else {
        panic!("the handshake is never answered, so the connection cannot complete")
    };
    server.abort();
    assert!(
        matches!(error.kind(), ErrorKind::Timeout(TimeoutKind::Connect)),
        "a connection that never completes is a connect timeout: {error:?}"
    );

    let (addr, server) = silent_server(true).await?;
    let mut config: Config = format!("redis://{addr}").into_config()?;
    config.command_timeout = Duration::from_millis(200);
    let client = Client::connect(config).await?;
    let error = client
        .get::<String>("key")
        .await
        .expect_err("the command is never answered");
    server.abort();
    assert!(
        matches!(error.kind(), ErrorKind::Timeout(TimeoutKind::Command)),
        "a command that never gets its reply is a command timeout: {error:?}"
    );

    Ok(())
}

/// `Unexpected` reported a dozen distinguishable conditions as
/// `client error: Unexpected error`, which tells the reader nothing and points
/// nowhere. Each one now names itself.
#[test]
fn an_internal_failure_names_the_condition() {
    for error in [
        ClientError::MalformedFrame,
        ClientError::InconsistentRespTape,
        ClientError::NotACollection,
        ClientError::MissingTransactionReply,
        ClientError::IncompatibleShardReplies,
        ClientError::NotAUnitVariant,
        ClientError::MissingMapValue,
    ] {
        let message = error.to_string();
        assert_ne!(
            "Unexpected error", message,
            "{error:?} must say what happened"
        );
        assert!(!message.is_empty());
    }
}

/// The frame parser raised `Unexpected`, which the framing list did not carry,
/// so a failure that leaves the reader at an unknown offset would have been
/// dispatched to a single caller with the stream possibly desynchronised.
#[test]
fn a_framing_failure_belongs_to_the_connection() {
    assert!(client(ClientError::MalformedFrame).is_connection_error());

    // Everything raised past framing still fails one command only.
    assert!(!client(ClientError::InconsistentRespTape).is_connection_error());
    assert!(!client(ClientError::NotACollection).is_connection_error());
    assert!(!client(ClientError::MissingTransactionReply).is_connection_error());
    assert!(!client(ClientError::IncompatibleShardReplies).is_connection_error());
}

/// The server error is the one class the application can act on by name — a
/// `WRONGTYPE` is a bug in the calling code, a `NOAUTH` a bug in the config.
#[test]
fn a_server_error_is_a_reply_the_server_chose_to_send() {
    assert!(redis(RedisErrorKind::WrongType).is_server_error());
    assert!(redis(RedisErrorKind::Other).is_server_error());

    assert!(!io().is_server_error());
    assert!(!client(ClientError::CannotParseInteger).is_server_error());
}

/// What a caller wanting to replay a command needs, in one predicate: the
/// transient failures, whatever layer they came from.
#[test]
fn a_retryable_error_covers_every_transient_layer() {
    assert!(io().is_retryable());
    assert!(Error::from(ErrorKind::EOF).is_retryable());
    assert!(timeout(TimeoutKind::Command).is_retryable());
    assert!(timeout(TimeoutKind::Connect).is_retryable());
    assert!(redis(RedisErrorKind::TryAgain).is_retryable());
    assert!(redis(RedisErrorKind::ClusterDown).is_retryable());
    assert!(redis(RedisErrorKind::MasterDown).is_retryable());
    assert!(redis(RedisErrorKind::NoMasterLink).is_retryable());
    assert!(redis(RedisErrorKind::Loading).is_retryable());
    assert!(redis(RedisErrorKind::Busy).is_retryable());

    // Replaying these produces the very same error.
    assert!(!redis(RedisErrorKind::WrongType).is_retryable());
    assert!(!redis(RedisErrorKind::NoAuth).is_retryable());
    assert!(!redis(RedisErrorKind::Err).is_retryable());
    assert!(!client(ClientError::MismatchedKeySlots).is_retryable());
    assert!(!Error::from(ErrorKind::Aborted).is_retryable());
}

/// A server restarting with persistence answers `LOADING` until its dataset is
/// in memory, and one running a long script answers `BUSY`. Both are outages,
/// not defects of the command, so they must not fall into `Other`, where
/// `is_retryable` cannot see them.
#[test]
fn loading_and_busy_are_parsed_and_retryable() -> Result<()> {
    let loading =
        RedisError::try_from(b"LOADING Redis is loading the dataset in memory".as_slice())?;
    assert_eq!(RedisErrorKind::Loading, loading.kind);
    assert_eq!(
        "LOADING Redis is loading the dataset in memory",
        loading.to_string()
    );
    assert!(Error::from(loading).is_retryable());

    let busy = RedisError::try_from(
        b"BUSY Redis is busy running a script. You can only call SCRIPT KILL or SHUTDOWN NOSAVE."
            .as_slice(),
    )?;
    assert_eq!(RedisErrorKind::Busy, busy.kind);
    assert!(busy.to_string().starts_with("BUSY Redis is busy"));
    assert!(Error::from(busy).is_retryable());

    Ok(())
}

/// The codes are matched whole, so the codes that only start with `BUSY` keep
/// their own meaning. `BUSYKEY` is what `RESTORE` answers on an existing key:
/// a defect of the command, which a replay only repeats.
#[test]
fn codes_starting_with_busy_are_not_busy() -> Result<()> {
    let busy_group =
        RedisError::try_from(b"BUSYGROUP Consumer Group name already exists".as_slice())?;
    assert_eq!(RedisErrorKind::BusyGroup, busy_group.kind);
    assert!(!Error::from(busy_group).is_retryable());

    let busy_key = RedisError::try_from(b"BUSYKEY Target key name already exists.".as_slice())?;
    assert_eq!(RedisErrorKind::Other, busy_key.kind);
    assert_eq!(
        "BUSYKEY Target key name already exists.",
        busy_key.to_string()
    );
    assert!(!Error::from(busy_key).is_retryable());

    Ok(())
}

/// A server message is bytes, not text. Reading it as text is lossy, so the
/// bytes stay reachable: a key or an argument the server echoed back is
/// recoverable byte for byte, whatever it holds.
#[test]
fn a_non_utf8_server_message_keeps_its_bytes() -> Result<()> {
    let raw: &[u8] = b"ERR unknown command '\xff\xfe'";
    let error = RedisError::try_from(raw)?;

    assert_eq!(b"unknown command '\xff\xfe'", error.description_bytes());
    assert!(error.description().contains('\u{fffd}'));
    assert_eq!("unknown command '��'", error.description());

    Ok(())
}

/// A rendered error holds no space it did not earn. An unrecognised kind renders
/// as nothing and a redirection carries no message of its own, so the separator
/// has to be dropped on both sides rather than written unconditionally.
#[test]
fn a_rendered_error_has_no_stray_space() -> Result<()> {
    let unclassified = RedisError::try_from(b"SOMEWEIRD failure text".as_slice())?;
    assert_eq!("SOMEWEIRD failure text", unclassified.to_string());
    assert_eq!(
        "redis server error: SOMEWEIRD failure text",
        Error::from(ErrorKind::Redis(unclassified)).to_string()
    );

    let moved = RedisError::try_from(b"MOVED 3999 127.0.0.1:6381".as_slice())?;
    assert_eq!("MOVED 3999 127.0.0.1:6381", moved.to_string());

    // The ordinary shape, kind and message both present, is unchanged.
    let wrong_type = RedisError::try_from(b"WRONGTYPE Operation against a key".as_slice())?;
    assert_eq!("WRONGTYPE Operation against a key", wrong_type.to_string());

    Ok(())
}