featherbit/plugins/util/
store_kv.rs

1//! Shared plumbing for the `store-*` nodes.
2//!
3//! Store resolution happens once, at policy-compile time: the node holds the
4//! resulting handle and nothing reads the registry on the request path. This is
5//! the pattern `limit-count` established (`src/plugins/native/limit_count.rs`).
6
7use std::collections::HashMap;
8use std::sync::Arc;
9
10use bytes::Bytes;
11use serde_json::Value;
12
13use crate::context::{Context, GatewayError};
14use crate::plugins::resources::PluginResources;
15use crate::plugins::PluginExecutionError;
16use crate::vars::template::Template;
17
18/// A resolved store, held by a node for its lifetime.
19///
20/// `RedisStoreClient` carries its own hand-written `Debug` impl (its
21/// connection manager has none), so `Arc<RedisStoreClient>` is `Debug` too and
22/// this can derive rather than hand-write. The four `store-*` plugins that
23/// hold one still can't derive their own `Debug`: their other fields are only
24/// read inside the `#[cfg(feature = "redis-store")]` `execute` body, and a
25/// headless build has no such body to read them, so a derived impl (which
26/// the dead-code pass ignores) would leave them looking unused there.
27#[derive(Debug)]
28pub struct StoreHandle {
29    /// Declared `stores:` name, carried for error messages.
30    pub name: String,
31    #[cfg(feature = "redis-store")]
32    client: Arc<crate::stores::redis_store::RedisStoreClient>,
33}
34
35impl StoreHandle {
36    /// A pooled connection to the store.
37    #[cfg(feature = "redis-store")]
38    pub async fn conn(&self) -> Result<crate::stores::redis_store::StoreConn, String> {
39        self.client.conn().await
40    }
41
42    /// The namespaced key this node should operate on.
43    #[cfg(feature = "redis-store")]
44    pub fn key_for(&self, rendered: &str) -> String {
45        namespaced_key(self.client.key_prefix(), rendered)
46    }
47}
48
49/// Builds the full redis key for a rendered policy key.
50///
51/// The `kv:` segment keeps policy-written keys from colliding with the `cnt:`,
52/// session and `acme:` keys that share the same store.
53#[cfg(feature = "redis-store")]
54pub fn namespaced_key(prefix: &str, rendered: &str) -> String {
55    format!(
56        "{}:{}:{}",
57        prefix,
58        crate::stores::namespaces::POLICY_KV,
59        rendered
60    )
61}
62
63/// Prepares the response common to every store-node error before it exits
64/// the node's `error` port: status + JSON body + `content-type`, the same
65/// shape `limit-count` and the session plugins already use (e.g.
66/// `authz_casdoor::store_error`, `limit_count.rs`'s `RATE_LIMIT_UNAVAILABLE`
67/// branch).
68///
69/// Without this, a policy that wires `error -> client` -- structurally the
70/// only alternative to leaving the port unwired, and the port model nudges
71/// every port toward being wired -- would answer with whatever
72/// `Context::new` left in `ctx.response` (`status_code == 0`, mapped to `200`
73/// by the listener) instead of a real failure status. That is exactly the
74/// fail-open the design's §7 rules out.
75fn prepare_error_response(ctx: &mut Context, status: u16, code: &str, message: &str) {
76    ctx.response.status_code = status;
77    ctx.response.body =
78        Bytes::from(serde_json::json!({ "error": code, "message": message }).to_string());
79    ctx.response.headers.insert(
80        "content-type".to_string(),
81        vec!["application/json".to_string()],
82    );
83}
84
85/// A `key` template that rendered to nothing.
86///
87/// `{prefix}:kv:` is still inside the namespace, so this is not a collision --
88/// but it would put every request on one shared key, turning a template typo
89/// into a cross-tenant leak. It fails loudly instead, with a `500`: this is a
90/// policy-authoring/config fault, not an outage.
91///
92/// Only ever checked on the redis-backed data path (rendering a key is
93/// pointless without a backend to read or write it against), so this is
94/// `#[cfg(feature = "redis-store")]` like the rest of that path.
95#[cfg(feature = "redis-store")]
96pub fn key_invalid(
97    mut ctx: Context,
98    node_type: &str,
99    op: &str,
100    store: &str,
101) -> PluginExecutionError {
102    let message = format!(
103        "{}: the 'key' template rendered to an empty string",
104        node_type
105    );
106    prepare_error_response(&mut ctx, 500, "STORE_KEY_INVALID", &message);
107    let mut metadata = HashMap::new();
108    metadata.insert("store".to_string(), Value::String(store.to_string()));
109    metadata.insert("op".to_string(), Value::String(op.to_string()));
110    PluginExecutionError {
111        context: ctx,
112        error: GatewayError {
113            node_id: String::new(),
114            code: "STORE_KEY_INVALID".to_string(),
115            message,
116            metadata,
117        },
118    }
119}
120
121/// Resolves the `store` config key to a handle, at construction time.
122#[cfg(feature = "redis-store")]
123pub fn resolve(
124    config: &HashMap<String, Value>,
125    resources: &Arc<PluginResources>,
126    node_type: &str,
127) -> Result<StoreHandle, String> {
128    let name = config
129        .get("store")
130        .and_then(|v| v.as_str())
131        .filter(|s| !s.is_empty())
132        .ok_or_else(|| {
133            format!(
134                "{}: 'store' is required and must name a declared stores: entry",
135                node_type
136            )
137        })?;
138    let client = resources.stores.load().client(name)?;
139    Ok(StoreHandle {
140        name: name.to_string(),
141        client,
142    })
143}
144
145/// Without the `redis-store` feature there is no backend to resolve, so the
146/// node fails at policy-compile time with a message that names the reason
147/// rather than failing mysteriously at request time.
148#[cfg(not(feature = "redis-store"))]
149pub fn resolve(
150    config: &HashMap<String, Value>,
151    _resources: &Arc<PluginResources>,
152    node_type: &str,
153) -> Result<StoreHandle, String> {
154    let name = config
155        .get("store")
156        .and_then(|v| v.as_str())
157        .filter(|s| !s.is_empty())
158        .ok_or_else(|| {
159            format!(
160                "{}: 'store' is required and must name a declared stores: entry",
161                node_type
162            )
163        })?;
164    Err(format!(
165        "{}: store '{}': this binary was built without the redis-store feature",
166        node_type, name
167    ))
168}
169
170/// Parses a required, templated string field.
171pub fn required_template(
172    config: &HashMap<String, Value>,
173    field: &str,
174    node_type: &str,
175) -> Result<Template, String> {
176    let raw = config
177        .get(field)
178        .and_then(|v| v.as_str())
179        .filter(|s| !s.is_empty())
180        .ok_or_else(|| format!("{}: '{}' is required", node_type, field))?;
181    let (tpl, warnings) = Template::parse(raw);
182    for w in warnings {
183        tracing::warn!(node_type = %node_type, field = %field, "{}", w);
184    }
185    Ok(tpl)
186}
187
188/// Parses the optional `ttl_seconds`. Absent means no expiry; `0` is rejected.
189///
190/// `store-get` (Task 2, the first caller of this module) had no use for a
191/// TTL -- reading a key never sets one -- so this was test-only for a while.
192/// `store-set` now calls it unconditionally from `from_config`, same as
193/// `required_template`: this is pure config validation with no dependency on
194/// a store actually being reachable, so unlike `resolve` it needs no
195/// `redis-store`-gated pair -- one definition, always compiled, matches how
196/// it is called on every feature combination.
197pub fn optional_ttl(
198    config: &HashMap<String, Value>,
199    node_type: &str,
200) -> Result<Option<u64>, String> {
201    optional_seconds(config, "ttl_seconds", node_type)
202}
203
204/// Parses an optional positive duration in seconds.
205///
206/// Absent means "no duration"; `0` is rejected rather than read as "none",
207/// because the two are too easy to confuse and omitting the field is how you
208/// say it. Pure config parsing with no store dependency, so unlike `resolve`
209/// this needs no `redis-store`-gated pair.
210pub fn optional_seconds(
211    config: &HashMap<String, Value>,
212    field: &str,
213    node_type: &str,
214) -> Result<Option<u64>, String> {
215    match config.get(field) {
216        None | Some(Value::Null) => Ok(None),
217        Some(v) => {
218            let n = v.as_u64().ok_or_else(|| {
219                format!("{}: '{}' must be a non-negative integer", node_type, field)
220            })?;
221            if n == 0 {
222                return Err(format!(
223                    "{}: '{}' must be greater than 0; omit the field to leave it unset",
224                    node_type, field
225                ));
226            }
227            Ok(Some(n))
228        }
229    }
230}
231
232/// A store outage. Never conflated with a miss: see the spec's §7. Prepares a
233/// `503` -- the dependency is unavailable, not the caller's fault -- so a
234/// policy that wires `error -> client` cannot fail open with a `200`.
235///
236/// Not feature-gated: the `#[cfg(not(feature = "redis-store"))]` `execute`
237/// bodies in all four plugins call this too, to report "built without the
238/// redis-store feature" with the same shape.
239pub fn store_error(
240    mut ctx: Context,
241    node_type: &str,
242    op: &str,
243    store: &str,
244    msg: String,
245) -> PluginExecutionError {
246    let message = format!("{}: {} failed: {}", node_type, op, msg);
247    prepare_error_response(&mut ctx, 503, "STORE_ERROR", &message);
248    let mut metadata = HashMap::new();
249    metadata.insert("store".to_string(), Value::String(store.to_string()));
250    metadata.insert("op".to_string(), Value::String(op.to_string()));
251    PluginExecutionError {
252        context: ctx,
253        error: GatewayError {
254            node_id: String::new(),
255            code: "STORE_ERROR".to_string(),
256            message,
257            metadata,
258        },
259    }
260}
261
262/// A value that exists but is not usable as configured (bad JSON, non-numeric).
263/// Prepares a `500`: a data/config fault, not an outage.
264///
265/// Only ever raised on the redis-backed data path -- see [`key_invalid`].
266#[cfg(feature = "redis-store")]
267pub fn value_invalid(
268    mut ctx: Context,
269    node_type: &str,
270    op: &str,
271    store: &str,
272    msg: String,
273) -> PluginExecutionError {
274    let message = format!("{}: {}", node_type, msg);
275    prepare_error_response(&mut ctx, 500, "STORE_VALUE_INVALID", &message);
276    let mut metadata = HashMap::new();
277    metadata.insert("store".to_string(), Value::String(store.to_string()));
278    metadata.insert("op".to_string(), Value::String(op.to_string()));
279    PluginExecutionError {
280        context: ctx,
281        error: GatewayError {
282            node_id: String::new(),
283            code: "STORE_VALUE_INVALID".to_string(),
284            message,
285            metadata,
286        },
287    }
288}
289
290/// True when a redis error reflects a value/type problem rather than an
291/// outage: a command applied to a key holding the wrong type (`WRONGTYPE`,
292/// surfaced by the crate as an `ExtensionError` with that code), or a numeric
293/// operation against a non-numeric value (`ERR value is not an integer or
294/// out of range`). Both are data faults `store-incr` reports as
295/// `STORE_VALUE_INVALID`, never `STORE_ERROR`.
296///
297/// Classifying on `code()`/`kind()` rather than matching the whole message
298/// avoids depending on redis wrapping the message text a particular way; the
299/// substring check is kept only for the one case (`ResponseError`'s "not an
300/// integer" wording) that has no dedicated code of its own.
301#[cfg(feature = "redis-store")]
302pub fn is_value_type_error(e: &redis::RedisError) -> bool {
303    e.code() == Some("WRONGTYPE")
304        || (e.kind() == redis::ErrorKind::ResponseError && e.to_string().contains("not an integer"))
305}
306
307#[cfg(all(test, feature = "redis-store"))]
308mod tests {
309    use super::*;
310    use crate::context::{GatewayRequest, Protocol};
311    use std::collections::HashMap;
312
313    fn cfg(json: serde_json::Value) -> HashMap<String, serde_json::Value> {
314        serde_json::from_value(json).unwrap()
315    }
316
317    fn test_ctx() -> Context {
318        Context::new(GatewayRequest {
319            method: "GET".to_string(),
320            path: "/".to_string(),
321            host: "example.com".to_string(),
322            scheme: "http".to_string(),
323            headers: HashMap::new(),
324            query_params: HashMap::new(),
325            body: bytes::Bytes::new(),
326            remote_addr: "10.1.2.3:44321".to_string(),
327            protocol: Protocol::Http1,
328        })
329    }
330
331    /// Keys are namespaced so a policy cannot collide with session, counter or
332    /// ACME keys in a store all of them share.
333    #[test]
334    fn test_key_for_namespaces_under_kv() {
335        assert_eq!(namespaced_key("fb", "retry:abc"), "fb:kv:retry:abc");
336    }
337
338    /// `key` is templated, so its rendered value can contain anything a caller
339    /// can put in a header. The rendered part is a *suffix*, so it cannot walk
340    /// back out of the namespace -- assert that against hostile inputs rather
341    /// than assuming it.
342    #[test]
343    fn test_a_rendered_key_cannot_escape_the_kv_namespace() {
344        let hostile = [
345            ":sess:{abc}",
346            "../sess:{abc}",
347            "fb:sess:{abc}",
348            "fb:acme:account",
349            "a\nfb:cnt:0:u1",
350        ];
351        for h in hostile {
352            let built = namespaced_key("fb", h);
353            assert!(
354                built.starts_with("fb:kv:"),
355                "escaped the namespace: {built}"
356            );
357            for ns in crate::stores::namespaces::MANAGED {
358                assert!(
359                    !built.starts_with(&format!("fb:{ns}:")),
360                    "{built} lands in the {ns} namespace"
361                );
362            }
363        }
364    }
365
366    /// An empty rendered key is not a collision -- `fb:kv:` is still inside the
367    /// namespace -- but it silently puts every request on one shared key, so a
368    /// template typo becomes a cross-tenant leak. Fail loudly instead, with a
369    /// `500` (a config/authoring fault, not an outage) and a non-empty body.
370    #[test]
371    fn test_empty_key_has_its_own_code() {
372        let e = key_invalid(test_ctx(), "store-get", "GET", "sessions");
373        assert_eq!(e.error.code, "STORE_KEY_INVALID");
374        assert_eq!(e.context.response.status_code, 500);
375        assert!(!e.context.response.body.is_empty());
376        assert_eq!(e.error.metadata.get("store").unwrap(), "sessions");
377        assert_eq!(e.error.metadata.get("op").unwrap(), "GET");
378    }
379
380    #[test]
381    fn test_required_template_rejects_a_missing_key() {
382        let err = required_template(&cfg(serde_json::json!({})), "key", "store-get").unwrap_err();
383        assert!(
384            err.contains("store-get"),
385            "error must name the node type: {err}"
386        );
387        assert!(err.contains("key"), "error must name the field: {err}");
388    }
389
390    #[test]
391    fn test_required_template_parses_a_template() {
392        let t = required_template(
393            &cfg(serde_json::json!({ "key": "retry:{{request.path}}" })),
394            "key",
395            "store-get",
396        )
397        .unwrap();
398        assert!(!t.is_literal());
399    }
400
401    /// `0` is a config error, not "no expiry": the two readings are too easy to
402    /// confuse, and omitting the field is how you say "no expiry".
403    #[test]
404    fn test_optional_ttl_rejects_zero() {
405        let err =
406            optional_ttl(&cfg(serde_json::json!({ "ttl_seconds": 0 })), "store-set").unwrap_err();
407        assert!(err.contains("ttl_seconds"), "{err}");
408    }
409
410    #[test]
411    fn test_optional_ttl_absent_is_none_and_present_is_some() {
412        assert_eq!(
413            optional_ttl(&cfg(serde_json::json!({})), "store-set").unwrap(),
414            None
415        );
416        assert_eq!(
417            optional_ttl(&cfg(serde_json::json!({ "ttl_seconds": 300 })), "store-set").unwrap(),
418            Some(300)
419        );
420    }
421
422    /// A store outage must be identifiable downstream, so the code is fixed and
423    /// the metadata carries enough to debug it. It must also leave the
424    /// response a real `503` with a non-empty body -- never the `200` an
425    /// untouched `Context::new` would answer with if a policy wires
426    /// `error -> client` -- which is the failure this whole fix round exists
427    /// to close (spec §7: never fail open).
428    #[test]
429    fn test_store_error_prepares_a_503_and_carries_code_store_and_op() {
430        let e = store_error(
431            test_ctx(),
432            "store-get",
433            "GET",
434            "sessions",
435            "connection refused".to_string(),
436        );
437        assert_eq!(e.error.code, "STORE_ERROR");
438        assert_eq!(e.error.metadata.get("store").unwrap(), "sessions");
439        assert_eq!(e.error.metadata.get("op").unwrap(), "GET");
440        assert!(
441            e.error.message.contains("connection refused"),
442            "{}",
443            e.error.message
444        );
445        assert_eq!(e.context.response.status_code, 503);
446        assert!(!e.context.response.body.is_empty());
447        assert_eq!(
448            e.context.response.headers.get("content-type").unwrap(),
449            &vec!["application/json".to_string()]
450        );
451    }
452
453    /// Same as above for `value_invalid`, whose status is `500`: a value/JSON
454    /// fault is a data problem, not an outage.
455    #[test]
456    fn test_value_invalid_prepares_a_500_and_carries_code_store_and_op() {
457        let e = value_invalid(
458            test_ctx(),
459            "store-get",
460            "GET",
461            "sessions",
462            "expected JSON".to_string(),
463        );
464        assert_eq!(e.error.code, "STORE_VALUE_INVALID");
465        assert_eq!(e.error.metadata.get("store").unwrap(), "sessions");
466        assert_eq!(e.error.metadata.get("op").unwrap(), "GET");
467        assert_eq!(e.context.response.status_code, 500);
468        assert!(!e.context.response.body.is_empty());
469    }
470
471    /// `WRONGTYPE` (a key holding a list/hash) and the "not an integer"
472    /// message (a key holding a non-numeric string) must both classify as a
473    /// value-type error, not an outage -- see `store-incr`'s docs and the
474    /// spec's §5.3.
475    #[test]
476    fn test_is_value_type_error_covers_wrongtype_and_not_an_integer() {
477        // `make_extension_error` is how the redis crate itself builds an
478        // error whose `code()` is a raw RESP error code it does not have a
479        // dedicated `ErrorKind` for -- exactly WRONGTYPE's situation.
480        let wrongtype = redis::make_extension_error(
481            "WRONGTYPE".to_string(),
482            Some("Operation against a key holding the wrong kind of value".to_string()),
483        );
484        assert!(is_value_type_error(&wrongtype));
485
486        let not_an_integer = redis::RedisError::from((
487            redis::ErrorKind::ResponseError,
488            "value is not an integer or out of range",
489        ));
490        assert!(is_value_type_error(&not_an_integer));
491
492        let outage = redis::RedisError::from((redis::ErrorKind::IoError, "connection refused"));
493        assert!(!is_value_type_error(&outage));
494    }
495}
496
497/// Integration tests against a real redis, gated on `FEATHERBIT_TEST_REDIS_URL`
498/// -- the same env var CI's `redis:7`/`valkey:8` service-container matrix sets,
499/// and the same skip-if-unset pattern as `src/acme/live_tests.rs` and
500/// `src/stores/redis_store.rs::tests::test_ping_live`.
501///
502/// `PluginResources::empty()`, used by every unit test in the four `store-*`
503/// plugins, has no declared stores, so none of those tests can exercise a
504/// round trip against a backend -- that is what this module is for.
505#[cfg(all(test, feature = "redis-store"))]
506mod live_tests {
507    use super::*;
508    use crate::config::StoreConfig;
509    use crate::context::{Context, GatewayRequest, Protocol};
510    use crate::plugins::native::store_delete::StoreDeletePlugin;
511    use crate::plugins::native::store_get::StoreGetPlugin;
512    use crate::plugins::native::store_incr::StoreIncrPlugin;
513    use crate::plugins::native::store_set::StoreSetPlugin;
514    use crate::plugins::Plugin;
515    use crate::stores::StoreRegistry;
516    use std::collections::HashMap;
517    use std::time::Duration;
518
519    /// Every key this module writes -- through a plugin or directly via a raw
520    /// redis command -- gets this TTL, so a panicking or failing test still
521    /// self-cleans rather than leaking a permanent key onto a long-lived
522    /// redis. 60s is ample for these tests to run and observe the key.
523    const LIVE_TEST_TTL_SECONDS: u64 = 60;
524
525    /// Skips unless a real store is configured, the same gate the session and
526    /// ACME live tests use.
527    fn store_url() -> Option<String> {
528        std::env::var("FEATHERBIT_TEST_REDIS_URL")
529            .ok()
530            .filter(|s| !s.is_empty())
531    }
532
533    /// Builds `PluginResources` with one declared store named "test" pointing
534    /// at `url`.
535    fn resources_with_store(url: &str) -> Arc<PluginResources> {
536        let store_cfg: StoreConfig =
537            serde_yaml::from_str(&format!("name: test\ntype: redis\nurl: {url}\n"))
538                .expect("store config parses");
539        let registry = StoreRegistry::rebuild(&StoreRegistry::default(), &[store_cfg], None)
540            .expect("store registry builds against a reachable url");
541        let resources = PluginResources::empty();
542        resources.stores.store(Arc::new(registry));
543        resources
544    }
545
546    fn cfg(json: serde_json::Value) -> HashMap<String, serde_json::Value> {
547        serde_json::from_value(json).unwrap()
548    }
549
550    fn test_ctx() -> Context {
551        Context::new(GatewayRequest {
552            method: "GET".to_string(),
553            path: "/".to_string(),
554            host: "example.com".to_string(),
555            scheme: "http".to_string(),
556            headers: HashMap::new(),
557            query_params: HashMap::new(),
558            body: bytes::Bytes::new(),
559            remote_addr: "10.1.2.3:44321".to_string(),
560            protocol: Protocol::Http1,
561        })
562    }
563
564    /// A key unique to this run, so parallel tests in this module -- and
565    /// leftover data from a previous, possibly crashed run against the same
566    /// redis -- never collide with each other.
567    fn unique_key(label: &str) -> String {
568        format!("task6:{}:{}", label, uuid::Uuid::new_v4())
569    }
570
571    #[tokio::test]
572    async fn test_set_then_get_round_trips() {
573        let Some(url) = store_url() else {
574            eprintln!("skipping test_set_then_get_round_trips: FEATHERBIT_TEST_REDIS_URL not set");
575            return;
576        };
577        let resources = resources_with_store(&url);
578        let key = unique_key("roundtrip");
579
580        let set = StoreSetPlugin::from_config(
581            &cfg(serde_json::json!({ "store": "test", "key": key, "value": "1", "ttl_seconds": LIVE_TEST_TTL_SECONDS })),
582            &resources,
583        )
584        .unwrap();
585        let out = set.execute(test_ctx()).await.unwrap();
586        assert_eq!(out.port, None);
587
588        let get = StoreGetPlugin::from_config(
589            &cfg(serde_json::json!({ "store": "test", "key": key, "name": "v" })),
590            &resources,
591        )
592        .unwrap();
593        let out = get.execute(test_ctx()).await.unwrap();
594        assert_eq!(out.port, None);
595        assert_eq!(out.context.message.get("v").unwrap(), "1");
596    }
597
598    #[tokio::test]
599    async fn test_get_on_an_absent_key_exits_miss() {
600        let Some(url) = store_url() else {
601            eprintln!(
602                "skipping test_get_on_an_absent_key_exits_miss: FEATHERBIT_TEST_REDIS_URL not set"
603            );
604            return;
605        };
606        let resources = resources_with_store(&url);
607        let key = unique_key("absent");
608
609        let get = StoreGetPlugin::from_config(
610            &cfg(serde_json::json!({ "store": "test", "key": key, "name": "v" })),
611            &resources,
612        )
613        .unwrap();
614        let out = get.execute(test_ctx()).await.unwrap();
615        assert_eq!(out.port, Some("miss"));
616    }
617
618    #[tokio::test]
619    async fn test_json_true_flattens_an_object() {
620        let Some(url) = store_url() else {
621            eprintln!(
622                "skipping test_json_true_flattens_an_object: FEATHERBIT_TEST_REDIS_URL not set"
623            );
624            return;
625        };
626        let resources = resources_with_store(&url);
627        let key = unique_key("json");
628
629        let set = StoreSetPlugin::from_config(
630            &cfg(serde_json::json!({
631                "store": "test",
632                "key": key,
633                "value": r#"{"tier":"gold"}"#,
634                "ttl_seconds": LIVE_TEST_TTL_SECONDS,
635            })),
636            &resources,
637        )
638        .unwrap();
639        set.execute(test_ctx()).await.unwrap();
640
641        let get = StoreGetPlugin::from_config(
642            &cfg(serde_json::json!({
643                "store": "test",
644                "key": key,
645                "name": "profile",
646                "json": true,
647            })),
648            &resources,
649        )
650        .unwrap();
651        let out = get.execute(test_ctx()).await.unwrap();
652        assert_eq!(out.context.message.get("profile.tier").unwrap(), "gold");
653    }
654
655    #[tokio::test]
656    async fn test_json_true_on_invalid_json_exits_error() {
657        let Some(url) = store_url() else {
658            eprintln!(
659                "skipping test_json_true_on_invalid_json_exits_error: FEATHERBIT_TEST_REDIS_URL not set"
660            );
661            return;
662        };
663        let resources = resources_with_store(&url);
664        let key = unique_key("badjson");
665
666        let set = StoreSetPlugin::from_config(
667            &cfg(serde_json::json!({ "store": "test", "key": key, "value": "not json", "ttl_seconds": LIVE_TEST_TTL_SECONDS })),
668            &resources,
669        )
670        .unwrap();
671        set.execute(test_ctx()).await.unwrap();
672
673        let get = StoreGetPlugin::from_config(
674            &cfg(serde_json::json!({
675                "store": "test",
676                "key": key,
677                "name": "profile",
678                "json": true,
679            })),
680            &resources,
681        )
682        .unwrap();
683        let err = get.execute(test_ctx()).await.unwrap_err();
684        assert_eq!(err.error.code, "STORE_VALUE_INVALID");
685    }
686
687    #[tokio::test]
688    async fn test_delete_on_an_absent_key_succeeds() {
689        let Some(url) = store_url() else {
690            eprintln!(
691                "skipping test_delete_on_an_absent_key_succeeds: FEATHERBIT_TEST_REDIS_URL not set"
692            );
693            return;
694        };
695        let resources = resources_with_store(&url);
696        let key = unique_key("delete-absent");
697
698        let delete = StoreDeletePlugin::from_config(
699            &cfg(serde_json::json!({ "store": "test", "key": key })),
700            &resources,
701        )
702        .unwrap();
703        let out = delete.execute(test_ctx()).await.unwrap();
704        assert_eq!(out.port, None);
705    }
706
707    /// The behavior store-incr exists for, and the one most likely to
708    /// regress: the TTL is applied when the key is created and must never be
709    /// refreshed by a later increment, or a client that keeps retrying would
710    /// keep its own counter alive and the bound it exists to enforce would
711    /// never reset.
712    #[tokio::test]
713    async fn test_incr_does_not_refresh_the_ttl() {
714        let Some(url) = store_url() else {
715            eprintln!(
716                "skipping test_incr_does_not_refresh_the_ttl: FEATHERBIT_TEST_REDIS_URL not set"
717            );
718            return;
719        };
720        let resources = resources_with_store(&url);
721        let key = unique_key("ttl");
722        let redis_key = namespaced_key("fb", &key);
723
724        let client = resources.stores.load().client("test").unwrap();
725        let mut raw = client.conn().await.unwrap();
726
727        let incr = StoreIncrPlugin::from_config(
728            &cfg(serde_json::json!({
729                "store": "test",
730                "key": key,
731                "name": "n",
732                "ttl_seconds": LIVE_TEST_TTL_SECONDS,
733            })),
734            &resources,
735        )
736        .unwrap();
737
738        incr.execute(test_ctx()).await.unwrap();
739        let ttl1: i64 = redis::cmd("PTTL")
740            .arg(&redis_key)
741            .query_async(&mut raw)
742            .await
743            .unwrap();
744        assert!(ttl1 > 0, "key must have a TTL right after creation: {ttl1}");
745
746        tokio::time::sleep(Duration::from_millis(1100)).await;
747
748        incr.execute(test_ctx()).await.unwrap();
749        let ttl2: i64 = redis::cmd("PTTL")
750            .arg(&redis_key)
751            .query_async(&mut raw)
752            .await
753            .unwrap();
754
755        // A weaker `ttl2 < ttl1` does not discriminate: even an unconditional
756        // refresh resets the TTL to ~60000ms both times, and the two round
757        // trips' own timing jitter alone makes `ttl2 < ttl1` about as likely
758        // as not -- that assertion passed against a mutated, always-refresh
759        // script in review. Require the drop to be close to the full sleep
760        // instead: on the correct (guarded) script the TTL just keeps
761        // counting down, so ttl1 - ttl2 tracks the ~1100ms sleep; on a
762        // refreshing script it is reset back near 60000ms both times, so the
763        // drop is near zero.
764        let drop = ttl1 - ttl2;
765        assert!(
766            drop >= 900,
767            "TTL must keep counting down by about the sleep duration, not be refreshed by a later increment: ttl1={ttl1} ttl2={ttl2} drop={drop}"
768        );
769    }
770
771    /// A store outage must exit `error`, never `miss` -- the distinction the
772    /// design turns on (see the module doc comment on
773    /// `src/plugins/native/store_get.rs`).
774    ///
775    /// This does **not** point the store at a closed port, even though that
776    /// was the brief's suggested mechanism. Empirically (measured on this
777    /// box with `zz_diag_*` throwaway probes, since removed) it does not
778    /// exercise this path quickly: `store-get` reaches the backend through
779    /// `RedisStoreClient::conn()` (`src/stores/redis_store.rs`), which calls
780    /// `redis::aio::ConnectionManager::new_with_config` with only
781    /// `connection_timeout`/`response_timeout` overridden. `number_of_retries`
782    /// (6), `factor` (100) and `max_delay` are left at the redis crate's /
783    /// `backon`'s defaults, which back off up to a *jittered, 60-second-capped*
784    /// delay between each of 6 retries on the very first connection failure --
785    /// worst case minutes before `new_with_config` gives up and returns `Err`.
786    /// `connect_timeout_ms`, the only knob `StoreConfig` exposes, bounds one
787    /// attempt, not the backoff between retries, so it cannot shorten this. A
788    /// raw TCP connect and a plain (non-`ConnectionManager`) redis connection
789    /// both fail in ~2s against the same closed port -- confirming the delay
790    /// is this retry/backoff policy, not the OS or this machine. That is a
791    /// real, currently-live gap between the design's promised fast `503` and
792    /// actual behavior on a fresh connection failure, worth its own follow-up;
793    /// fixing it is out of this task's scope (`store_kv.rs` and
794    /// `E2E_TESTBOOK.md` only), so this test does not wait on it.
795    ///
796    /// Instead it reaches the *same* `Err` arm in `store-get`'s `execute`
797    /// (`conn.get(&key).await` failing) a different way that needs no new
798    /// connection at all: writing a non-string value directly (bypassing
799    /// `store-set`, which only ever writes strings) so `GET` fails with
800    /// redis's own `WRONGTYPE` error against an already-healthy connection.
801    /// `store-get` does not distinguish "the connection is down" from "the
802    /// command itself errored" -- both are `Err(e)` from the one `conn.get()`
803    /// call, mapped to the same `STORE_ERROR` by the same line of code -- so
804    /// this exercises exactly the branch a real outage would take,
805    /// deterministically and in milliseconds.
806    #[tokio::test]
807    async fn test_unreachable_store_exits_error_not_miss() {
808        let Some(url) = store_url() else {
809            eprintln!(
810                "skipping test_unreachable_store_exits_error_not_miss: FEATHERBIT_TEST_REDIS_URL not set"
811            );
812            return;
813        };
814        let resources = resources_with_store(&url);
815        let key = unique_key("backend-error");
816        let redis_key = namespaced_key("fb", &key);
817
818        // A list value: GET against it is a real backend error (WRONGTYPE),
819        // not a missing key.
820        let client = resources.stores.load().client("test").unwrap();
821        let mut raw = client.conn().await.unwrap();
822        let _: i64 = redis::cmd("LPUSH")
823            .arg(&redis_key)
824            .arg("x")
825            .query_async(&mut raw)
826            .await
827            .unwrap();
828        // Written directly (bypassing store-set), so it needs its own expiry
829        // to self-clean -- see `LIVE_TEST_TTL_SECONDS`.
830        let _: bool = redis::cmd("EXPIRE")
831            .arg(&redis_key)
832            .arg(LIVE_TEST_TTL_SECONDS)
833            .query_async(&mut raw)
834            .await
835            .unwrap();
836
837        let get = StoreGetPlugin::from_config(
838            &cfg(serde_json::json!({ "store": "test", "key": key, "name": "v" })),
839            &resources,
840        )
841        .unwrap();
842        let err = get.execute(test_ctx()).await.unwrap_err();
843        assert_eq!(err.error.code, "STORE_ERROR");
844    }
845
846    /// `store-incr` against a key holding a non-numeric string: the docs and
847    /// the spec both promise `STORE_VALUE_INVALID` for this, and it is the
848    /// case the substring match on "not an integer" was originally written
849    /// for -- see F6 in the final review.
850    #[tokio::test]
851    async fn test_incr_on_a_non_numeric_string_exits_value_invalid() {
852        let Some(url) = store_url() else {
853            eprintln!(
854                "skipping test_incr_on_a_non_numeric_string_exits_value_invalid: FEATHERBIT_TEST_REDIS_URL not set"
855            );
856            return;
857        };
858        let resources = resources_with_store(&url);
859        let key = unique_key("incr-non-numeric");
860
861        let set = StoreSetPlugin::from_config(
862            &cfg(serde_json::json!({ "store": "test", "key": key, "value": "not-a-number", "ttl_seconds": LIVE_TEST_TTL_SECONDS })),
863            &resources,
864        )
865        .unwrap();
866        set.execute(test_ctx()).await.unwrap();
867
868        let incr = StoreIncrPlugin::from_config(
869            &cfg(serde_json::json!({ "store": "test", "key": key, "name": "n" })),
870            &resources,
871        )
872        .unwrap();
873        let err = incr.execute(test_ctx()).await.unwrap_err();
874        assert_eq!(err.error.code, "STORE_VALUE_INVALID");
875    }
876
877    /// `store-incr` against a key holding a list (`WRONGTYPE`): the substring
878    /// match on "not an integer" missed this entirely, silently reporting a
879    /// pure data fault as `STORE_ERROR` -- an outage the caller did not have.
880    /// See F6 in the final review.
881    #[tokio::test]
882    async fn test_incr_on_a_list_value_exits_value_invalid_not_store_error() {
883        let Some(url) = store_url() else {
884            eprintln!(
885                "skipping test_incr_on_a_list_value_exits_value_invalid_not_store_error: FEATHERBIT_TEST_REDIS_URL not set"
886            );
887            return;
888        };
889        let resources = resources_with_store(&url);
890        let key = unique_key("incr-wrongtype");
891        let redis_key = namespaced_key("fb", &key);
892
893        let client = resources.stores.load().client("test").unwrap();
894        let mut raw = client.conn().await.unwrap();
895        let _: i64 = redis::cmd("LPUSH")
896            .arg(&redis_key)
897            .arg("x")
898            .query_async(&mut raw)
899            .await
900            .unwrap();
901        let _: bool = redis::cmd("EXPIRE")
902            .arg(&redis_key)
903            .arg(LIVE_TEST_TTL_SECONDS)
904            .query_async(&mut raw)
905            .await
906            .unwrap();
907
908        let incr = StoreIncrPlugin::from_config(
909            &cfg(serde_json::json!({ "store": "test", "key": key, "name": "n" })),
910            &resources,
911        )
912        .unwrap();
913        let err = incr.execute(test_ctx()).await.unwrap_err();
914        assert_eq!(err.error.code, "STORE_VALUE_INVALID");
915    }
916
917    /// The empty-key guard, actually executed rather than only asserted on
918    /// the constructor's constant (F7 in the final review): a template that
919    /// renders to nothing must exit `STORE_KEY_INVALID` with a `500`, not
920    /// silently operate on `{prefix}:kv:`.
921    #[tokio::test]
922    async fn test_empty_rendered_key_exits_store_key_invalid() {
923        let Some(url) = store_url() else {
924            eprintln!(
925                "skipping test_empty_rendered_key_exits_store_key_invalid: FEATHERBIT_TEST_REDIS_URL not set"
926            );
927            return;
928        };
929        let resources = resources_with_store(&url);
930
931        let get = StoreGetPlugin::from_config(
932            &cfg(serde_json::json!({ "store": "test", "key": "{{request.headers.x-absent}}", "name": "v" })),
933            &resources,
934        )
935        .unwrap();
936        let err = get.execute(test_ctx()).await.unwrap_err();
937        assert_eq!(err.error.code, "STORE_KEY_INVALID");
938        assert_eq!(err.context.response.status_code, 500);
939        assert!(!err.context.response.body.is_empty());
940    }
941
942    /// The one behaviour `reads_response_body()` exists for: a `key`/`value`
943    /// referencing the response body must force the route onto the buffered
944    /// path. `PluginResources::empty()` (used by every unit test in the four
945    /// plugins) has no declared stores, so `from_config` cannot succeed there
946    /// and the four unit tests could only assert `Template::
947    /// references_response_body()` on the side -- which cannot fail even if
948    /// `reads_response_body()` itself is hardcoded wrong (F3/F4 in the final
949    /// review). This constructs the real plugins against a declared store
950    /// and asserts the trait method directly.
951    #[tokio::test]
952    async fn test_reads_response_body_reflects_a_response_body_reference() {
953        let Some(url) = store_url() else {
954            eprintln!(
955                "skipping test_reads_response_body_reflects_a_response_body_reference: FEATHERBIT_TEST_REDIS_URL not set"
956            );
957            return;
958        };
959        let resources = resources_with_store(&url);
960
961        let get_plain = StoreGetPlugin::from_config(
962            &cfg(serde_json::json!({ "store": "test", "key": "k", "name": "n" })),
963            &resources,
964        )
965        .unwrap();
966        let get_reads = StoreGetPlugin::from_config(
967            &cfg(serde_json::json!({ "store": "test", "key": "{{response.body}}", "name": "n" })),
968            &resources,
969        )
970        .unwrap();
971        assert!(!get_plain.reads_response_body());
972        assert!(get_reads.reads_response_body());
973
974        let set_plain = StoreSetPlugin::from_config(
975            &cfg(serde_json::json!({ "store": "test", "key": "k", "value": "1" })),
976            &resources,
977        )
978        .unwrap();
979        let set_reads = StoreSetPlugin::from_config(
980            &cfg(serde_json::json!({ "store": "test", "key": "k", "value": "{{response.body}}" })),
981            &resources,
982        )
983        .unwrap();
984        assert!(!set_plain.reads_response_body());
985        assert!(set_reads.reads_response_body());
986
987        let delete_plain = StoreDeletePlugin::from_config(
988            &cfg(serde_json::json!({ "store": "test", "key": "k" })),
989            &resources,
990        )
991        .unwrap();
992        let delete_reads = StoreDeletePlugin::from_config(
993            &cfg(serde_json::json!({ "store": "test", "key": "{{response.body}}" })),
994            &resources,
995        )
996        .unwrap();
997        assert!(!delete_plain.reads_response_body());
998        assert!(delete_reads.reads_response_body());
999
1000        let incr_plain = StoreIncrPlugin::from_config(
1001            &cfg(serde_json::json!({ "store": "test", "key": "k", "name": "n" })),
1002            &resources,
1003        )
1004        .unwrap();
1005        let incr_reads = StoreIncrPlugin::from_config(
1006            &cfg(serde_json::json!({ "store": "test", "key": "{{response.body}}", "name": "n" })),
1007            &resources,
1008        )
1009        .unwrap();
1010        assert!(!incr_plain.reads_response_body());
1011        assert!(incr_reads.reads_response_body());
1012    }
1013
1014    /// The mirror of `test_incr_does_not_refresh_the_ttl`: with
1015    /// `refresh_ttl: true` the expiry is pushed back out on every increment,
1016    /// giving "N events within `ttl_seconds` of each other" instead of
1017    /// "N events since the first one".
1018    ///
1019    /// The assertion is the inverse of the guarded case, and deliberately as
1020    /// strong: a *rise* of nearly the whole sleep, not merely `ttl2 > ttl1`,
1021    /// which round-trip jitter alone could satisfy.
1022    #[tokio::test]
1023    async fn test_incr_with_refresh_ttl_extends_the_expiry() {
1024        let Some(url) = store_url() else {
1025            eprintln!(
1026                "skipping test_incr_with_refresh_ttl_extends_the_expiry: FEATHERBIT_TEST_REDIS_URL not set"
1027            );
1028            return;
1029        };
1030        let resources = resources_with_store(&url);
1031        let key = unique_key("ttl-slide");
1032        let redis_key = namespaced_key("fb", &key);
1033
1034        let client = resources.stores.load().client("test").unwrap();
1035        let mut raw = client.conn().await.unwrap();
1036
1037        let incr = StoreIncrPlugin::from_config(
1038            &cfg(serde_json::json!({
1039                "store": "test",
1040                "key": key,
1041                "name": "n",
1042                "ttl_seconds": LIVE_TEST_TTL_SECONDS,
1043                "refresh_ttl": true,
1044            })),
1045            &resources,
1046        )
1047        .unwrap();
1048
1049        incr.execute(test_ctx()).await.unwrap();
1050        let ttl1: i64 = redis::cmd("PTTL")
1051            .arg(&redis_key)
1052            .query_async(&mut raw)
1053            .await
1054            .unwrap();
1055        assert!(ttl1 > 0, "key must have a TTL right after creation: {ttl1}");
1056
1057        tokio::time::sleep(Duration::from_millis(1100)).await;
1058
1059        incr.execute(test_ctx()).await.unwrap();
1060        let ttl2: i64 = redis::cmd("PTTL")
1061            .arg(&redis_key)
1062            .query_async(&mut raw)
1063            .await
1064            .unwrap();
1065
1066        // Both samples are taken immediately after an increment, so a
1067        // refreshing script leaves them roughly EQUAL -- it is the *absence*
1068        // of the countdown that proves the refresh, not a rise. The guarded
1069        // (create-only) script would show a drop tracking the ~1100ms sleep,
1070        // exactly as `test_incr_does_not_refresh_the_ttl` asserts, so this
1071        // bound of 200ms fails against it while tolerating round-trip jitter.
1072        let drop = ttl1 - ttl2;
1073        assert!(
1074            drop <= 200,
1075            "refresh_ttl must re-arm the expiry on each increment, so it must not count down: ttl1={ttl1} ttl2={ttl2} drop={drop}"
1076        );
1077    }
1078
1079    /// `extend_ttl_seconds` makes a read push the key's expiry out, so an
1080    /// entry stays alive while it is being used and disappears a fixed time
1081    /// after the last access.
1082    #[tokio::test]
1083    async fn test_get_with_extend_ttl_pushes_the_expiry_out() {
1084        let Some(url) = store_url() else {
1085            eprintln!(
1086                "skipping test_get_with_extend_ttl_pushes_the_expiry_out: FEATHERBIT_TEST_REDIS_URL not set"
1087            );
1088            return;
1089        };
1090        let resources = resources_with_store(&url);
1091        let key = unique_key("touch");
1092        let redis_key = namespaced_key("fb", &key);
1093
1094        let client = resources.stores.load().client("test").unwrap();
1095        let mut raw = client.conn().await.unwrap();
1096
1097        let set = StoreSetPlugin::from_config(
1098            &cfg(serde_json::json!({
1099                "store": "test",
1100                "key": key,
1101                "value": "alive",
1102                "ttl_seconds": LIVE_TEST_TTL_SECONDS,
1103            })),
1104            &resources,
1105        )
1106        .unwrap();
1107        set.execute(test_ctx()).await.unwrap();
1108
1109        let ttl1: i64 = redis::cmd("PTTL")
1110            .arg(&redis_key)
1111            .query_async(&mut raw)
1112            .await
1113            .unwrap();
1114
1115        tokio::time::sleep(Duration::from_millis(1100)).await;
1116
1117        let get = StoreGetPlugin::from_config(
1118            &cfg(serde_json::json!({
1119                "store": "test",
1120                "key": key,
1121                "name": "v",
1122                "extend_ttl_seconds": LIVE_TEST_TTL_SECONDS,
1123            })),
1124            &resources,
1125        )
1126        .unwrap();
1127        let out = get.execute(test_ctx()).await.unwrap();
1128
1129        // The read still returns the value -- extending must not replace GET's job.
1130        assert_eq!(
1131            out.context.message.get("v").and_then(|v| v.as_str()),
1132            Some("alive")
1133        );
1134
1135        let ttl2: i64 = redis::cmd("PTTL")
1136            .arg(&redis_key)
1137            .query_async(&mut raw)
1138            .await
1139            .unwrap();
1140
1141        // Same shape as the refresh_ttl assertion above: ttl1 is sampled just
1142        // after the write and ttl2 just after the extending read, so a working
1143        // GETEX leaves them roughly equal. A plain GET would let the TTL count
1144        // down by the ~1100ms sleep, which this bound rejects.
1145        let drop = ttl1 - ttl2;
1146        assert!(
1147            drop <= 200,
1148            "a read with extend_ttl_seconds must re-arm the expiry, so it must not count down: ttl1={ttl1} ttl2={ttl2} drop={drop}"
1149        );
1150    }
1151
1152    /// Extending must not conjure a key. `GETEX` on a missing key returns nil
1153    /// and creates nothing, so the node still exits `miss` and the store is
1154    /// left untouched -- otherwise a keep-alive read would manufacture the
1155    /// very entries it is meant to keep warm.
1156    #[tokio::test]
1157    async fn test_get_with_extend_ttl_on_an_absent_key_still_misses() {
1158        let Some(url) = store_url() else {
1159            eprintln!(
1160                "skipping test_get_with_extend_ttl_on_an_absent_key_still_misses: FEATHERBIT_TEST_REDIS_URL not set"
1161            );
1162            return;
1163        };
1164        let resources = resources_with_store(&url);
1165        let key = unique_key("touch-absent");
1166        let redis_key = namespaced_key("fb", &key);
1167
1168        let client = resources.stores.load().client("test").unwrap();
1169        let mut raw = client.conn().await.unwrap();
1170
1171        let get = StoreGetPlugin::from_config(
1172            &cfg(serde_json::json!({
1173                "store": "test",
1174                "key": key,
1175                "name": "v",
1176                "extend_ttl_seconds": LIVE_TEST_TTL_SECONDS,
1177            })),
1178            &resources,
1179        )
1180        .unwrap();
1181
1182        let out = get.execute(test_ctx()).await.unwrap();
1183        assert_eq!(out.port, Some("miss"));
1184
1185        let exists: i64 = redis::cmd("EXISTS")
1186            .arg(&redis_key)
1187            .query_async(&mut raw)
1188            .await
1189            .unwrap();
1190        assert_eq!(exists, 0, "a missing key must not be created by extending");
1191    }
1192}
featherbit/plugins/util/store_kv.rs

featherbit/plugins/util/
store_kv.rs
