dialcache/
redis.rs

1//! Production [`Remote`] adapter over the [`redis`] crate.
2//!
3//! [`RedisAdapter`] borrows a connected, caller-owned connection handle: a
4//! [`ConnectionManager`] for standalone or Sentinel-managed servers, a
5//! [`ClusterConnection`] for Redis Cluster, or a plain
6//! [`MultiplexedConnection`]. It never connects, reconnects, drains or closes
7//! the handle and never changes its options. The caller remains responsible
8//! for:
9//!
10//! - **Finite budgets.** Configure connection, response and retry limits on
11//!   the handle (`ConnectionManagerConfig::set_response_timeout`,
12//!   `ClusterClientBuilder::response_timeout`, ...). The cache bounds its own
13//!   wait for reads and honors cooperative cancellation, but cancellation
14//!   after dispatch cannot prove a command did not execute.
15//! - **Primaries for tracked reads.** On a cluster handle the adapter routes
16//!   every tracked `MGET` explicitly to the slot primary, so replica lag can
17//!   never hide an invalidation watermark even when the caller enabled
18//!   `read_from_replicas`. A standalone or Sentinel-managed handle must point
19//!   at the primary itself; the adapter cannot tell a replica apart.
20//! - **Watermark retention.** Invalidation stores watermarks with a
21//!   retention of at least two hours that only widens. Configure eviction so
22//!   watermark keys are not evicted before tracked values age out; an
23//!   evicted watermark resurrects fenced values.
24//!
25//! Wire behavior matches the TypeScript and Go adapters exactly: an
26//! untracked read is one `GET`; a tracked read is one `MGET value watermark`
27//! on the primary; a write is exactly one native `SET key frame PX ttl`; an
28//! invalidation is `EVALSHA` of [`INVALIDATION_SCRIPT`] with one `EVAL`
29//! retry carrying identical arguments after any `EVALSHA` failure. Accepted
30//! but invalid replies are [`RedisProtocolError`]s and are never retried.
31
32use std::fmt;
33
34use futures::future::{self, BoxFuture, Either, FutureExt};
35use redis::aio::{ConnectionLike, ConnectionManager, MultiplexedConnection};
36use redis::cluster_async::ClusterConnection;
37use redis::cluster_routing::{Route, RoutingInfo, SingleNodeRoutingInfo, Slot, SlotAddr};
38use redis::{Cmd, RedisResult, Value};
39use sha1::{Digest, Sha1};
40
41use crate::error::BoxError;
42use crate::limits::{MAX_SAFE_INTEGER, MAX_SUPPORTED_DURATION_MS};
43use crate::protocol::{decode_frame, encode_frame, ProtocolError};
44use crate::remote::{
45    InvalidateRequest, ReadContext, ReadRequest, ReadResult, Remote, WriteRequest,
46};
47
48/// The version-1 wire invalidation transition, byte-identical to the Go
49/// `InvalidationScript` (a unit test pins this against `go/redis_adapter.go`).
50/// TypeScript's `INVALIDATE_CACHE_SCRIPT` is the same Lua with different
51/// whitespace and a comment, so its `EVALSHA` digest differs; every port
52/// self-heals through the `EVAL` fallback, so mixed-language clusters need no
53/// shared script cache.
54///
55/// Redis Lua numbers exactly represent the accepted safe-integer domain.
56/// Invalid arguments return before `GET`/repair; wrong-type keys alone are
57/// repairable read failures.
58pub const INVALIDATION_SCRIPT: &str = r#"local function parse_safe_integer(raw)
59  if not string.match(raw, "^%d+$") then return nil end
60  local value = tonumber(raw)
61  if not value or value > 9007199254740991 then return nil end
62  return value
63end
64local future_buffer_ms = parse_safe_integer(ARGV[1])
65if not future_buffer_ms or future_buffer_ms < 0 or future_buffer_ms > 31536000000 then
66  return redis.error_reply("ERR invalid DialCache future buffer")
67end
68local invalidated_at_ms = parse_safe_integer(ARGV[2])
69if not invalidated_at_ms or invalidated_at_ms > 9007199254740991 - future_buffer_ms then
70  return redis.error_reply("ERR invalid DialCache invalidatedAtMs")
71end
72local proposed_watermark = invalidated_at_ms + future_buffer_ms
73local raw_watermark = redis.pcall("GET", KEYS[1])
74if type(raw_watermark) == "table" and raw_watermark.err then
75  if not string.match(raw_watermark.err, "^WRONGTYPE ") then return raw_watermark end
76  raw_watermark = false
77end
78local current_watermark = 0
79if raw_watermark then
80  local parsed_watermark = parse_safe_integer(raw_watermark)
81  if parsed_watermark then current_watermark = parsed_watermark end
82end
83local watermark = math.max(current_watermark, proposed_watermark)
84local current_ttl_ms = -2
85if raw_watermark then current_ttl_ms = redis.call("PTTL", KEYS[1]) end
86local desired_ttl_ms = math.max(7200000, watermark - invalidated_at_ms + 3600000 + 60000)
87if current_ttl_ms > desired_ttl_ms then desired_ttl_ms = current_ttl_ms end
88local encoded_watermark = string.format("%.0f", watermark)
89if current_ttl_ms == -1 then
90  redis.call("SET", KEYS[1], encoded_watermark)
91else
92  redis.call("SET", KEYS[1], encoded_watermark, "PX", desired_ttl_ms)
93end
94return 1"#;
95
96/// Lowercase hex SHA-1 of [`INVALIDATION_SCRIPT`], the `EVALSHA` argument.
97pub fn invalidation_script_sha1() -> String {
98    hex::encode(Sha1::digest(INVALIDATION_SCRIPT.as_bytes()))
99}
100
101/// A reply the server accepted but that violates the DialCache wire protocol.
102///
103/// Such replies are never retried: the command executed and the adapter has
104/// no way to tell what state the server holds.
105#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
106#[error("Invalid DialCache Redis reply: {message}")]
107pub struct RedisProtocolError {
108    /// What the reply violated: an unexpected reply type, a tracked read
109    /// without exactly two bulk values, a `SET` reply other than `OK`, or an
110    /// invalidation reply other than the integer `1`.
111    pub message: String,
112}
113
114impl RedisProtocolError {
115    fn new(message: impl Into<String>) -> Self {
116        RedisProtocolError {
117            message: message.into(),
118        }
119    }
120}
121
122/// A read stopped waiting because its [`ReadContext`] was cancelled.
123///
124/// The command may still execute on the server; the cache's own deadline is
125/// authoritative and this error only ends the adapter's wait.
126#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
127#[error("DialCache Redis read cancelled before a reply arrived")]
128pub struct RedisReadCancelled;
129
130/// A cloneable, task-shareable handle that runs one Redis command.
131///
132/// Implemented for [`ConnectionManager`], [`MultiplexedConnection`] and
133/// [`ClusterConnection`]. Test doubles implement it to observe dispatched
134/// arguments without a server. Every method borrows `&self`; implementations
135/// clone the underlying handle per command, which the `redis` crate's handles
136/// make cheap.
137pub trait RedisConnection: Send + Sync + 'static {
138    /// Run one command with the handle's default routing.
139    fn run(&self, cmd: Cmd) -> BoxFuture<'_, RedisResult<Value>>;
140
141    /// Run one command that reads `key`, on the node that owns the key's
142    /// slot as a primary. Non-cluster handles have exactly one node and use
143    /// their default routing.
144    fn run_on_primary(&self, key: &str, cmd: Cmd) -> BoxFuture<'_, RedisResult<Value>>;
145}
146
147/// Lift a RESP3 error reply carried as a value into the error channel so
148/// every caller sees one failure shape.
149fn settle(reply: RedisResult<Value>) -> RedisResult<Value> {
150    match reply {
151        Ok(Value::ServerError(error)) => Err(error.into()),
152        other => other,
153    }
154}
155
156fn run_cloned<C>(connection: &C, cmd: Cmd) -> BoxFuture<'_, RedisResult<Value>>
157where
158    C: ConnectionLike + Clone + Send + Sync,
159{
160    let mut connection = connection.clone();
161    async move { settle(connection.req_packed_command(&cmd).await) }.boxed()
162}
163
164impl RedisConnection for ConnectionManager {
165    fn run(&self, cmd: Cmd) -> BoxFuture<'_, RedisResult<Value>> {
166        run_cloned(self, cmd)
167    }
168
169    fn run_on_primary(&self, _key: &str, cmd: Cmd) -> BoxFuture<'_, RedisResult<Value>> {
170        run_cloned(self, cmd)
171    }
172}
173
174impl RedisConnection for MultiplexedConnection {
175    fn run(&self, cmd: Cmd) -> BoxFuture<'_, RedisResult<Value>> {
176        run_cloned(self, cmd)
177    }
178
179    fn run_on_primary(&self, _key: &str, cmd: Cmd) -> BoxFuture<'_, RedisResult<Value>> {
180        run_cloned(self, cmd)
181    }
182}
183
184impl RedisConnection for ClusterConnection {
185    fn run(&self, cmd: Cmd) -> BoxFuture<'_, RedisResult<Value>> {
186        run_cloned(self, cmd)
187    }
188
189    /// Selecting the slot primary is explicit here because a stale replica
190    /// could hide an invalidation fence when the client enabled replica reads.
191    fn run_on_primary(&self, key: &str, cmd: Cmd) -> BoxFuture<'_, RedisResult<Value>> {
192        let route = Route::with_slot(Slot::for_key(key), SlotAddr::Master);
193        let routing = RoutingInfo::SingleNode(SingleNodeRoutingInfo::SpecificNode(route));
194        let mut connection = self.clone();
195        async move { settle(connection.route_command(cmd, routing).await) }.boxed()
196    }
197}
198
199/// [`Remote`] over a caller-owned `redis` crate connection handle.
200///
201/// See the [module documentation](self) for the caller's responsibilities.
202#[derive(Clone)]
203pub struct RedisAdapter<C> {
204    connection: C,
205}
206
207impl<C> fmt::Debug for RedisAdapter<C> {
208    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
209        f.debug_struct("RedisAdapter").finish_non_exhaustive()
210    }
211}
212
213impl<C: RedisConnection> RedisAdapter<C> {
214    /// Borrow `connection`. The adapter never connects or closes it.
215    pub fn new(connection: C) -> Self {
216        RedisAdapter { connection }
217    }
218
219    /// The borrowed connection handle.
220    pub fn connection(&self) -> &C {
221        &self.connection
222    }
223
224    /// The script's raw decimal boundary for out-of-band tooling and
225    /// conformance replay. Validation inside Lua precedes every mutation.
226    ///
227    /// `EVALSHA` runs first; any failure is retried exactly once as `EVAL`
228    /// with identical arguments, and a failed retry surfaces unmodified. The
229    /// transition is idempotent (the watermark only advances and its
230    /// retention only widens), so a duplicate run after an ambiguous failure
231    /// is harmless. A successful reply must be the integer `1`.
232    pub fn invalidate_decimal<'a>(
233        &'a self,
234        watermark_key: &'a str,
235        future_buffer_ms: &'a str,
236        invalidated_at_ms: &'a str,
237    ) -> BoxFuture<'a, Result<(), BoxError>> {
238        async move {
239            let script_command = |program: &str, script: &str| {
240                let mut cmd = redis::cmd(program);
241                cmd.arg(script)
242                    .arg(1)
243                    .arg(watermark_key)
244                    .arg(future_buffer_ms)
245                    .arg(invalidated_at_ms);
246                cmd
247            };
248            let evalsha = script_command("EVALSHA", &invalidation_script_sha1());
249            let reply = match self.connection.run(evalsha).await {
250                Ok(reply) => reply,
251                Err(_) => {
252                    let eval = script_command("EVAL", INVALIDATION_SCRIPT);
253                    self.connection.run(eval).await?
254                }
255            };
256            validate_invalidation_reply(&reply)?;
257            Ok(())
258        }
259        .boxed()
260    }
261}
262
263/// One bulk-string reply position: `None` is absent.
264type Bulk = Option<Vec<u8>>;
265
266/// One bulk-string reply position: `Nil` is absent, string replies are
267/// content, anything else is a protocol violation.
268fn bulk(reply: Value) -> Result<Bulk, RedisProtocolError> {
269    match reply {
270        Value::Nil => Ok(None),
271        Value::BulkString(bytes) => Ok(Some(bytes)),
272        Value::SimpleString(text) => Ok(Some(text.into_bytes())),
273        other => Err(RedisProtocolError::new(format!(
274            "expected Redis bulk string, got {}",
275            value_kind(&other)
276        ))),
277    }
278}
279
280/// A tracked read must return exactly two bulk values.
281fn tracked_reply(reply: Value) -> Result<(Bulk, Bulk), RedisProtocolError> {
282    let Value::Array(items) = reply else {
283        return Err(RedisProtocolError::new(
284            "tracked read must return exactly two bulk values",
285        ));
286    };
287    let [value, watermark]: [Value; 2] = items
288        .try_into()
289        .map_err(|_| RedisProtocolError::new("tracked read must return exactly two bulk values"))?;
290    Ok((bulk(value)?, bulk(watermark)?))
291}
292
293/// `SET` must answer `OK`.
294pub fn validate_set_reply(reply: &Value) -> Result<(), RedisProtocolError> {
295    let ok = match reply {
296        Value::Okay => true,
297        Value::SimpleString(text) => text == "OK",
298        Value::BulkString(bytes) => bytes == b"OK",
299        _ => false,
300    };
301    if ok {
302        Ok(())
303    } else {
304        Err(RedisProtocolError::new(
305            "invalid Redis SET reply; expected OK",
306        ))
307    }
308}
309
310/// The invalidation script must answer the integer `1`.
311pub fn validate_invalidation_reply(reply: &Value) -> Result<(), RedisProtocolError> {
312    if matches!(reply, Value::Int(1)) {
313        Ok(())
314    } else {
315        Err(RedisProtocolError::new(
316            "invalid Redis invalidation reply; expected integer 1",
317        ))
318    }
319}
320
321fn value_kind(value: &Value) -> &'static str {
322    match value {
323        Value::Nil => "nil",
324        Value::Int(_) => "integer",
325        Value::BulkString(_) => "bulk string",
326        Value::Array(_) => "array",
327        Value::SimpleString(_) => "simple string",
328        Value::Okay => "OK",
329        Value::Map(_) => "map",
330        Value::Attribute { .. } => "attribute",
331        Value::Set(_) => "set",
332        Value::Double(_) => "double",
333        Value::Boolean(_) => "boolean",
334        Value::VerbatimString { .. } => "verbatim string",
335        Value::BigNumber(_) => "big number",
336        Value::Push { .. } => "push",
337        Value::ServerError(_) => "server error",
338        _ => "unknown",
339    }
340}
341
342impl<C: RedisConnection> Remote for RedisAdapter<C> {
343    fn read(
344        &self,
345        request: ReadRequest,
346        context: ReadContext,
347    ) -> BoxFuture<'_, Result<ReadResult, BoxError>> {
348        async move {
349            if context.cancel.is_cancelled() {
350                return Err(Box::new(RedisReadCancelled) as BoxError);
351            }
352            let tracked = request.watermark_key.is_some();
353            let command = match &request.watermark_key {
354                None => {
355                    let mut cmd = redis::cmd("GET");
356                    cmd.arg(&request.value_key);
357                    self.connection.run(cmd)
358                }
359                Some(watermark_key) => {
360                    let mut cmd = redis::cmd("MGET");
361                    cmd.arg(&request.value_key).arg(watermark_key);
362                    self.connection.run_on_primary(&request.value_key, cmd)
363                }
364            };
365            let reply = match future::select(command, context.cancel.cancelled()).await {
366                Either::Left((reply, _)) => reply?,
367                Either::Right(((), _)) => return Err(Box::new(RedisReadCancelled) as BoxError),
368            };
369            let (value, watermark) = if tracked {
370                tracked_reply(reply)?
371            } else {
372                (bulk(reply)?, None)
373            };
374            Ok(decode_frame(
375                value.as_deref(),
376                tracked,
377                watermark.as_deref(),
378            )?)
379        }
380        .boxed()
381    }
382
383    /// Exactly one native `SET` of the complete frame. It neither reads nor
384    /// modifies the entity watermark; the frame's stamp is stored exactly.
385    fn write(&self, request: WriteRequest) -> BoxFuture<'_, Result<(), BoxError>> {
386        async move {
387            if request.ttl_ms == 0 || request.ttl_ms > MAX_SUPPORTED_DURATION_MS {
388                return Err(ProtocolError::InvalidDuration.into());
389            }
390            let raw = encode_frame(&request.frame)?;
391            let mut cmd = redis::cmd("SET");
392            cmd.arg(&request.value_key)
393                .arg(raw)
394                .arg("PX")
395                .arg(request.ttl_ms.to_string());
396            let reply = self.connection.run(cmd).await?;
397            validate_set_reply(&reply)?;
398            Ok(())
399        }
400        .boxed()
401    }
402
403    /// The caller's clock sample stays stable through the `EVAL` retry, so
404    /// one logical invalidation proposes one watermark.
405    fn invalidate(&self, request: InvalidateRequest) -> BoxFuture<'_, Result<(), BoxError>> {
406        async move {
407            if request.future_buffer_ms > MAX_SUPPORTED_DURATION_MS {
408                return Err(ProtocolError::InvalidDuration.into());
409            }
410            if request.invalidated_at_ms > MAX_SAFE_INTEGER - request.future_buffer_ms {
411                return Err(ProtocolError::InvalidTimestamp.into());
412            }
413            self.invalidate_decimal(
414                &request.watermark_key,
415                &request.future_buffer_ms.to_string(),
416                &request.invalidated_at_ms.to_string(),
417            )
418            .await
419        }
420        .boxed()
421    }
422}
423
424#[cfg(test)]
425mod tests {
426    use super::*;
427    use crate::cancel::CancelToken;
428    use crate::codec::Payload;
429    use crate::remote::{Frame, MissReason};
430    use parking_lot::Mutex;
431    use std::collections::VecDeque;
432    use std::path::Path;
433    use std::sync::Arc;
434    use std::time::Duration;
435
436    /// One recorded dispatch: whether it was pinned to a primary, and its
437    /// arguments including the command name.
438    #[derive(Debug, Clone, PartialEq, Eq)]
439    struct Call {
440        primary: bool,
441        args: Vec<Vec<u8>>,
442    }
443
444    /// Replays scripted replies in order. A dispatch beyond the script fails
445    /// with a client error so a test that expected no dispatch fails loudly;
446    /// [`Scripted::hanging`] makes it wait forever instead.
447    #[derive(Default)]
448    struct Scripted {
449        calls: Mutex<Vec<Call>>,
450        replies: Mutex<VecDeque<RedisResult<Value>>>,
451        hang_when_exhausted: bool,
452    }
453
454    impl Scripted {
455        fn with(replies: Vec<RedisResult<Value>>) -> Arc<Self> {
456            Arc::new(Scripted {
457                calls: Mutex::new(Vec::new()),
458                replies: Mutex::new(replies.into_iter().collect()),
459                hang_when_exhausted: false,
460            })
461        }
462
463        /// A connection whose replies never arrive, for cancellation tests.
464        fn hanging() -> Arc<Self> {
465            Arc::new(Scripted {
466                calls: Mutex::new(Vec::new()),
467                replies: Mutex::new(VecDeque::new()),
468                hang_when_exhausted: true,
469            })
470        }
471
472        fn calls(&self) -> Vec<Call> {
473            self.calls.lock().clone()
474        }
475
476        fn dispatch(&self, primary: bool, cmd: &Cmd) -> BoxFuture<'static, RedisResult<Value>> {
477            let args = cmd
478                .args_iter()
479                .map(|arg| match arg {
480                    redis::Arg::Simple(bytes) => bytes.to_vec(),
481                    _ => b"<cursor>".to_vec(),
482                })
483                .collect();
484            self.calls.lock().push(Call { primary, args });
485            match self.replies.lock().pop_front() {
486                Some(reply) => future::ready(reply).boxed(),
487                // A dispatch beyond the script is a test failure, not a hang,
488                // unless the test asked for a reply that never arrives.
489                None if self.hang_when_exhausted => future::pending().boxed(),
490                None => future::ready(Err(redis::RedisError::from((
491                    redis::ErrorKind::Io,
492                    "unexpected dispatch beyond the scripted replies",
493                ))))
494                .boxed(),
495            }
496        }
497    }
498
499    impl RedisConnection for Arc<Scripted> {
500        fn run(&self, cmd: Cmd) -> BoxFuture<'_, RedisResult<Value>> {
501            self.dispatch(false, &cmd)
502        }
503
504        fn run_on_primary(&self, _key: &str, cmd: Cmd) -> BoxFuture<'_, RedisResult<Value>> {
505            self.dispatch(true, &cmd)
506        }
507    }
508
509    fn failure() -> RedisResult<Value> {
510        Err(redis::RedisError::from((
511            redis::ErrorKind::Io,
512            "ambiguous response loss",
513        )))
514    }
515
516    fn strings(args: &[Vec<u8>]) -> Vec<String> {
517        args.iter()
518            .map(|arg| String::from_utf8_lossy(arg).into_owned())
519            .collect()
520    }
521
522    fn block_on<T>(future: impl std::future::Future<Output = T>) -> T {
523        futures::executor::block_on(future)
524    }
525
526    fn tracked(value_key: &str, watermark_key: &str) -> ReadRequest {
527        ReadRequest {
528            value_key: value_key.to_string(),
529            watermark_key: Some(watermark_key.to_string()),
530        }
531    }
532
533    fn context() -> ReadContext {
534        ReadContext {
535            timeout_ms: 50,
536            cancel: CancelToken::new(),
537        }
538    }
539
540    fn go_invalidation_script() -> String {
541        let path = Path::new(env!("CARGO_MANIFEST_DIR")).join("../go/redis_adapter.go");
542        let source = std::fs::read_to_string(&path)
543            .unwrap_or_else(|error| panic!("read {}: {error}", path.display()));
544        let start = source
545            .find("const InvalidationScript = `")
546            .expect("Go declares InvalidationScript")
547            + "const InvalidationScript = `".len();
548        let end = source[start..]
549            .find('`')
550            .expect("Go raw string literal closes");
551        source[start..start + end].to_string()
552    }
553
554    #[test]
555    fn invalidation_script_matches_go_byte_for_byte() {
556        assert_eq!(INVALIDATION_SCRIPT, go_invalidation_script());
557    }
558
559    #[test]
560    fn invalidation_script_sha1_matches_independent_digests() {
561        let sha = invalidation_script_sha1();
562        // Pinned from `shasum -a 1` over the Go constant. TypeScript's
563        // INVALIDATE_CACHE_SCRIPT_SHA1 (e90603d37a1a3d7a49ae5417bea0d595b94ddec6)
564        // hashes the same Lua with different formatting.
565        assert_eq!(sha, "a6c1c661884bd7f535a13c79135a40b1edd2e216");
566        assert_eq!(sha, redis::Script::new(INVALIDATION_SCRIPT).get_hash());
567    }
568
569    #[test]
570    fn write_is_one_set_with_px_and_validates_ok() {
571        let frame = Frame {
572            created_at_ms: 1_700_000_000_000,
573            payload: Payload::binary(vec![0, 1, 255]),
574        };
575        let connection = Scripted::with(vec![Ok(Value::Okay)]);
576        let adapter = RedisAdapter::new(connection.clone());
577        block_on(adapter.write(WriteRequest {
578            value_key: "value".into(),
579            frame: frame.clone(),
580            ttl_ms: 2,
581        }))
582        .expect("SET succeeds");
583        let calls = connection.calls();
584        assert_eq!(calls.len(), 1);
585        assert!(!calls[0].primary);
586        assert_eq!(
587            calls[0].args,
588            vec![
589                b"SET".to_vec(),
590                b"value".to_vec(),
591                encode_frame(&frame).unwrap(),
592                b"PX".to_vec(),
593                b"2".to_vec(),
594            ]
595        );
596
597        for ttl_ms in [0, MAX_SUPPORTED_DURATION_MS + 1] {
598            let error = block_on(adapter.write(WriteRequest {
599                value_key: "value".into(),
600                frame: frame.clone(),
601                ttl_ms,
602            }))
603            .expect_err("invalid TTL rejected");
604            assert!(error.is::<ProtocolError>(), "{error}");
605        }
606        assert_eq!(connection.calls().len(), 1, "invalid input dispatched");
607
608        for reply in [
609            Value::Nil,
610            Value::SimpleString("ok".into()),
611            Value::Boolean(true),
612            Value::Int(1),
613        ] {
614            assert!(validate_set_reply(&reply).is_err(), "{reply:?}");
615        }
616        for reply in [
617            Value::Okay,
618            Value::SimpleString("OK".into()),
619            Value::BulkString(b"OK".to_vec()),
620        ] {
621            assert!(validate_set_reply(&reply).is_ok(), "{reply:?}");
622        }
623    }
624
625    #[test]
626    fn write_reports_protocol_error_on_bad_reply() {
627        let connection = Scripted::with(vec![Ok(Value::Int(1))]);
628        let adapter = RedisAdapter::new(connection);
629        let error = block_on(adapter.write(WriteRequest {
630            value_key: "value".into(),
631            frame: Frame {
632                created_at_ms: 1,
633                payload: Payload::text("1"),
634            },
635            ttl_ms: 1,
636        }))
637        .expect_err("bad reply rejected");
638        assert!(error.is::<RedisProtocolError>(), "{error}");
639    }
640
641    #[test]
642    fn invalidate_retries_evalsha_once_as_eval_with_identical_arguments() {
643        let connection = Scripted::with(vec![failure(), Ok(Value::Int(1))]);
644        let adapter = RedisAdapter::new(connection.clone());
645        block_on(adapter.invalidate(InvalidateRequest {
646            watermark_key: "watermark".into(),
647            invalidated_at_ms: 1_700_000_000_000,
648            future_buffer_ms: 100,
649        }))
650        .expect("EVAL retry succeeds");
651        let calls = connection.calls();
652        assert_eq!(calls.len(), 2, "{calls:?}");
653        let first = strings(&calls[0].args);
654        let second = strings(&calls[1].args);
655        assert_eq!(first[0], "EVALSHA");
656        assert_eq!(first[1], invalidation_script_sha1());
657        assert_eq!(second[0], "EVAL");
658        assert_eq!(second[1], INVALIDATION_SCRIPT);
659        assert_eq!(first[2..], second[2..], "retry changed logical arguments");
660        assert_eq!(
661            first[2..],
662            ["1", "watermark", "100", "1700000000000"].map(str::to_string)
663        );
664    }
665
666    #[test]
667    fn invalidate_failed_retry_surfaces_and_stops() {
668        let connection = Scripted::with(vec![failure(), failure()]);
669        let adapter = RedisAdapter::new(connection.clone());
670        let error = block_on(adapter.invalidate_decimal("watermark", "0", "1"))
671            .expect_err("second failure surfaces");
672        assert!(error.is::<redis::RedisError>(), "{error}");
673        assert_eq!(connection.calls().len(), 2);
674    }
675
676    #[test]
677    fn invalidate_accepted_invalid_reply_is_not_retried() {
678        let connection = Scripted::with(vec![Ok(Value::BulkString(b"1".to_vec()))]);
679        let adapter = RedisAdapter::new(connection.clone());
680        let error = block_on(adapter.invalidate(InvalidateRequest {
681            watermark_key: "watermark".into(),
682            invalidated_at_ms: 1,
683            future_buffer_ms: 0,
684        }))
685        .expect_err("string reply rejected");
686        assert!(error.is::<RedisProtocolError>(), "{error}");
687        assert_eq!(
688            connection.calls().len(),
689            1,
690            "invalid accepted reply retried"
691        );
692
693        for reply in [
694            Value::Nil,
695            Value::BulkString(Vec::new()),
696            Value::Int(0),
697            Value::Double(1.0),
698            Value::BulkString(b"1".to_vec()),
699            Value::SimpleString("1".into()),
700            Value::Boolean(true),
701            Value::Okay,
702        ] {
703            assert!(validate_invalidation_reply(&reply).is_err(), "{reply:?}");
704        }
705        assert!(validate_invalidation_reply(&Value::Int(1)).is_ok());
706    }
707
708    #[test]
709    fn invalidate_rejects_out_of_domain_arguments_before_dispatch() {
710        let connection = Scripted::with(vec![]);
711        let adapter = RedisAdapter::new(connection.clone());
712        for (invalidated_at_ms, future_buffer_ms) in [
713            (0, MAX_SUPPORTED_DURATION_MS + 1),
714            (MAX_SAFE_INTEGER, 1),
715            (MAX_SAFE_INTEGER + 1, 0),
716        ] {
717            let error = block_on(adapter.invalidate(InvalidateRequest {
718                watermark_key: "watermark".into(),
719                invalidated_at_ms,
720                future_buffer_ms,
721            }))
722            .expect_err("invalid domain rejected");
723            assert!(error.is::<ProtocolError>(), "{error}");
724        }
725        assert!(connection.calls().is_empty());
726    }
727
728    #[test]
729    fn tracked_read_is_one_primary_mget_classified_by_the_protocol() {
730        let raw = encode_frame(&Frame {
731            created_at_ms: 2,
732            payload: Payload::text("1"),
733        })
734        .unwrap();
735        let bulk = |bytes: &[u8]| Value::BulkString(bytes.to_vec());
736        struct Case {
737            reply: Value,
738            expected: Result<ReadResult, ()>,
739        }
740        let hit = || {
741            Ok(ReadResult::Hit(Frame {
742                created_at_ms: 2,
743                payload: Payload::text("1"),
744            }))
745        };
746        let miss = |reason, observed| {
747            Ok(ReadResult::Miss {
748                reason,
749                observed_watermark_ms: observed,
750            })
751        };
752        let cases = [
753            Case {
754                reply: Value::Array(vec![bulk(&raw), bulk(b"1")]),
755                expected: hit(),
756            },
757            Case {
758                reply: Value::Array(vec![bulk(&raw), bulk(b"2")]),
759                expected: miss(MissReason::WatermarkFenced, Some(2)),
760            },
761            Case {
762                reply: Value::Array(vec![Value::Nil, bulk(b"5")]),
763                expected: miss(MissReason::ValueAbsent, Some(5)),
764            },
765            Case {
766                reply: Value::Array(vec![bulk(&raw), Value::Nil]),
767                expected: hit(),
768            },
769            Case {
770                reply: Value::Array(vec![bulk(&raw), bulk(b"bad")]),
771                expected: miss(MissReason::Unclassified, None),
772            },
773            Case {
774                reply: Value::Array(vec![bulk(&raw)]),
775                expected: Err(()),
776            },
777            Case {
778                reply: Value::Array(vec![Value::Int(12), Value::Nil]),
779                expected: Err(()),
780            },
781            Case {
782                reply: Value::Array(vec![bulk(&raw), Value::Boolean(false)]),
783                expected: Err(()),
784            },
785            Case {
786                reply: bulk(&raw),
787                expected: Err(()),
788            },
789        ];
790        for case in cases {
791            let connection = Scripted::with(vec![Ok(case.reply.clone())]);
792            let adapter = RedisAdapter::new(connection.clone());
793            let got = block_on(adapter.read(tracked("value", "watermark"), context()));
794            match (&case.expected, &got) {
795                (Ok(expected), Ok(got)) => assert_eq!(got, expected, "{:?}", case.reply),
796                (Err(()), Err(error)) => {
797                    assert!(
798                        error.is::<RedisProtocolError>(),
799                        "{:?}: {error}",
800                        case.reply
801                    )
802                }
803                _ => panic!("reply {:?}: {got:?}", case.reply),
804            }
805            assert_eq!(
806                connection.calls(),
807                vec![Call {
808                    primary: true,
809                    args: vec![b"MGET".to_vec(), b"value".to_vec(), b"watermark".to_vec()],
810                }]
811            );
812        }
813    }
814
815    #[test]
816    fn untracked_read_is_one_get() {
817        let raw = encode_frame(&Frame {
818            created_at_ms: 7,
819            payload: Payload::binary(vec![0, 255]),
820        })
821        .unwrap();
822        let connection = Scripted::with(vec![Ok(Value::Nil), Ok(Value::BulkString(raw))]);
823        let adapter = RedisAdapter::new(connection.clone());
824        let request = ReadRequest {
825            value_key: "value".into(),
826            watermark_key: None,
827        };
828        assert_eq!(
829            block_on(adapter.read(request.clone(), context())).unwrap(),
830            ReadResult::miss(MissReason::ValueAbsent)
831        );
832        assert_eq!(
833            block_on(adapter.read(request, context())).unwrap(),
834            ReadResult::Hit(Frame {
835                created_at_ms: 7,
836                payload: Payload::binary(vec![0, 255]),
837            })
838        );
839        let calls = connection.calls();
840        assert_eq!(calls.len(), 2);
841        for call in calls {
842            assert_eq!(
843                call,
844                Call {
845                    primary: false,
846                    args: vec![b"GET".to_vec(), b"value".to_vec()],
847                }
848            );
849        }
850    }
851
852    #[test]
853    fn read_payload_encoding_error_is_an_error_not_a_miss() {
854        let mut raw = encode_frame(&Frame {
855            created_at_ms: 2,
856            payload: Payload::text("1"),
857        })
858        .unwrap();
859        raw[9] = 7;
860        let connection = Scripted::with(vec![Ok(Value::BulkString(raw))]);
861        let adapter = RedisAdapter::new(connection);
862        let error = block_on(adapter.read(
863            ReadRequest {
864                value_key: "value".into(),
865                watermark_key: None,
866            },
867            context(),
868        ))
869        .expect_err("unknown encoding tag is an error");
870        assert_eq!(
871            error.downcast_ref::<ProtocolError>(),
872            Some(&ProtocolError::PayloadEncoding)
873        );
874    }
875
876    #[test]
877    fn read_honors_cancellation_before_and_during_the_wait() {
878        let connection = Scripted::hanging();
879        let adapter = RedisAdapter::new(connection.clone());
880        let cancelled = ReadContext {
881            timeout_ms: 50,
882            cancel: CancelToken::new(),
883        };
884        cancelled.cancel.cancel();
885        let error = block_on(adapter.read(tracked("value", "watermark"), cancelled))
886            .expect_err("pre-cancelled read never dispatches");
887        assert!(error.is::<RedisReadCancelled>(), "{error}");
888        assert!(connection.calls().is_empty());
889
890        let context = context();
891        let token = context.cancel.clone();
892        let canceller = std::thread::spawn(move || {
893            std::thread::sleep(Duration::from_millis(20));
894            token.cancel();
895        });
896        let error = block_on(adapter.read(tracked("value", "watermark"), context))
897            .expect_err("hung command is abandoned on cancel");
898        canceller.join().unwrap();
899        assert!(error.is::<RedisReadCancelled>(), "{error}");
900        assert_eq!(
901            connection.calls().len(),
902            1,
903            "the command was dispatched once"
904        );
905    }
906}
dialcache/redis.rs

dialcache/
redis.rs
