featherbit/plugins/native/
proxy_cache.rs

1//! Response caching (`proxy-cache`), ported from APISIX's
2//! `apisix/plugins/proxy-cache`.
3//!
4//! Serving a cached response means *looking up* the cache before the upstream
5//! call and *storing* the fresh response after it — two moments a single
6//! featherbit node cannot span. The behavior is expressed as a **pair of
7//! nodes** sharing one cache, linked by a required `id`:
8//!
9//! - a **lookup** node placed *before* `upstream`, which serves a cache hit
10//!   straight to the client (short-circuiting the upstream call), and
11//! - a **store** node placed *after* `upstream`, which caches a fresh response
12//!   for later hits.
13//!
14//! Both nodes derive the cache key identically from the same `cache_key`
15//! template and the request, and share one namespace via `id`, so they always
16//! agree. State lives in [`crate::traffic::CacheRegistry`].
17//!
18//! # Wiring
19//!
20//! ```text
21//!            ┌──────────────────┐        ┌──────────┐        ┌──────────────────┐
22//!  listener →│ proxy-cache      │success →│ upstream │success →│ proxy-cache      │→ client
23//!            │  (phase=lookup)  │        │          │        │  (phase=store)   │
24//!            └──────────────────┘        └──────────┘        └──────────────────┘
25//!                    │ error                                    (caches responses
26//!                    ▼                                        whose status is cacheable)
27//!               client.in
28//!         (cached response, HIT)
29//! ```
30//!
31//! On a hit, the lookup node writes the cached response onto the context, adds
32//! `featherbit-cache-status: HIT`, and fails with `PROXY_CACHE_HIT` — its `error`
33//! port goes to `client.in`, delivering the cached response without touching
34//! the upstream. On a miss it passes through; the store node then caches the
35//! upstream response and marks it `featherbit-cache-status: MISS`.
36
37use async_trait::async_trait;
38use std::collections::HashMap;
39use std::sync::Arc;
40use std::time::Duration;
41
42use crate::context::{Context, GatewayError};
43use crate::plugins::resources::PluginResources;
44use crate::plugins::{Plugin, PluginExecutionError, PluginOutput, PluginResult};
45
46/// Header written by both nodes to report the cache outcome.
47const CACHE_STATUS_HEADER: &str = "featherbit-cache-status";
48/// Response headers hidden from clients when `hide_cache_headers` is set.
49const HIDDEN_HEADERS: &[&str] = &["cache-control", "expires"];
50
51/// Which half of the pair this node is.
52#[derive(Debug, Clone, Copy, PartialEq)]
53enum Role {
54    /// Runs before `upstream`: serves a cache hit.
55    Lookup,
56    /// Runs after `upstream`: stores a fresh response.
57    Store,
58}
59
60/// One node of a `proxy-cache` lookup/store pair.
61///
62/// Holds a handle to the process-wide [`crate::traffic::CacheRegistry`]; the
63/// key is derived per request from `cache_key`, namespaced by `id`.
64pub struct ProxyCachePlugin {
65    role: Role,
66    /// Shared cache namespace — links the lookup and store nodes.
67    id: String,
68    /// Cache-key components, each interpolated and joined per request.
69    cache_key: Vec<String>,
70    /// Freshness lifetime for stored entries.
71    cache_ttl: Duration,
72    /// Response statuses eligible for caching.
73    cache_statuses: Vec<u16>,
74    /// HTTP methods eligible for caching (uppercase).
75    cache_methods: Vec<String>,
76    /// When set, hides upstream cache headers from served cache hits.
77    hide_cache_headers: bool,
78    resources: Arc<PluginResources>,
79}
80
81impl ProxyCachePlugin {
82    /// Builds one node of the pair from node config.
83    ///
84    /// Accepted keys:
85    /// - `phase` / `role` (string, **required**): `lookup` (before upstream) or
86    ///   `store` (after upstream).
87    /// - `id` (string, **required**): shared cache namespace; the lookup and
88    ///   store nodes of one pair must use the same `id`.
89    /// - `cache_key` (array of string templates **or** a single string,
90    ///   default `["$request_method", "$host", "$uri"]`): components
91    ///   interpolated (see [`crate::vars::interpolate`]) and joined to form the
92    ///   key. Both nodes must configure it identically.
93    /// - `cache_ttl` (integer seconds, default `300`): freshness lifetime.
94    /// - `cache_http_statuses` (array, default `[200, 301, 404]`): statuses
95    ///   eligible for caching. (`cache_http_status`, APISIX's singular spelling,
96    ///   is also accepted.)
97    /// - `cache_method` (array, default `["GET", "HEAD"]`): cacheable methods.
98    /// - `hide_cache_headers` (bool, default `false`): strip `cache-control` /
99    ///   `expires` from served cache hits.
100    ///
101    /// ```yaml
102    /// # before upstream
103    /// type: proxy-cache
104    /// config:
105    ///   phase: lookup
106    ///   id: catalog
107    ///   cache_key: ["$request_method", "$host", "$uri"]
108    ///   cache_ttl: 300
109    /// ---
110    /// # after upstream
111    /// type: proxy-cache
112    /// config:
113    ///   phase: store
114    ///   id: catalog
115    ///   cache_key: ["$request_method", "$host", "$uri"]
116    ///   cache_ttl: 300
117    ///   cache_http_statuses: [200, 301, 404]
118    /// ```
119    pub fn from_config(
120        config: &HashMap<String, serde_json::Value>,
121        resources: &Arc<PluginResources>,
122    ) -> Result<Self, String> {
123        let role = match config
124            .get("phase")
125            .or_else(|| config.get("role"))
126            .and_then(|v| v.as_str())
127        {
128            Some("lookup") => Role::Lookup,
129            Some("store") => Role::Store,
130            Some(other) => {
131                return Err(format!(
132                    "proxy-cache: unknown phase/role '{}' (expected 'lookup' or 'store')",
133                    other
134                ))
135            }
136            None => {
137                return Err(
138                    "proxy-cache: 'phase' (or 'role') is required: 'lookup' or 'store'".to_string(),
139                )
140            }
141        };
142
143        let id = config
144            .get("id")
145            .and_then(|v| v.as_str())
146            .filter(|s| !s.trim().is_empty())
147            .ok_or("proxy-cache: 'id' is required (links the lookup/store pair)")?
148            .to_string();
149
150        let cache_key = match config.get("cache_key") {
151            None => vec![
152                "$request_method".to_string(),
153                "$host".to_string(),
154                "$uri".to_string(),
155            ],
156            Some(serde_json::Value::String(s)) => vec![s.clone()],
157            Some(serde_json::Value::Array(items)) => {
158                let mut out = Vec::with_capacity(items.len());
159                for item in items {
160                    let s = item
161                        .as_str()
162                        .ok_or("proxy-cache: cache_key entries must be strings")?;
163                    out.push(s.to_string());
164                }
165                if out.is_empty() {
166                    return Err("proxy-cache: cache_key must not be empty".to_string());
167                }
168                out
169            }
170            Some(_) => {
171                return Err(
172                    "proxy-cache: cache_key must be a string or an array of strings".to_string(),
173                )
174            }
175        };
176
177        let ttl_secs = config
178            .get("cache_ttl")
179            .and_then(|v| v.as_u64())
180            .unwrap_or(300);
181        if ttl_secs == 0 {
182            return Err("proxy-cache: cache_ttl must be >= 1 second".to_string());
183        }
184
185        let cache_statuses = parse_statuses(
186            config
187                .get("cache_http_statuses")
188                .or_else(|| config.get("cache_http_status")),
189        )?
190        .unwrap_or_else(|| vec![200, 301, 404]);
191
192        let cache_methods = match config.get("cache_method") {
193            None => vec!["GET".to_string(), "HEAD".to_string()],
194            Some(v) => {
195                let arr = v
196                    .as_array()
197                    .ok_or("proxy-cache: cache_method must be an array of strings")?;
198                let mut out = Vec::with_capacity(arr.len());
199                for item in arr {
200                    let m = item
201                        .as_str()
202                        .ok_or("proxy-cache: cache_method entries must be strings")?;
203                    out.push(m.to_uppercase());
204                }
205                if out.is_empty() {
206                    return Err("proxy-cache: cache_method must not be empty".to_string());
207                }
208                out
209            }
210        };
211
212        let hide_cache_headers = config
213            .get("hide_cache_headers")
214            .and_then(|v| v.as_bool())
215            .unwrap_or(false);
216
217        Ok(Self {
218            role,
219            id,
220            cache_key,
221            cache_ttl: Duration::from_secs(ttl_secs),
222            cache_statuses,
223            cache_methods,
224            hide_cache_headers,
225            resources: resources.clone(),
226        })
227    }
228
229    /// Whether this request's method is cacheable.
230    fn method_cacheable(&self, ctx: &Context) -> bool {
231        let method = ctx.request.method.to_uppercase();
232        self.cache_methods.contains(&method)
233    }
234
235    /// Derives the cache key: `id` namespace + interpolated `cache_key`
236    /// components joined by a control-char separator (outside the character set
237    /// of any header/method/path, so components can't collide).
238    fn derive_key(&self, ctx: &Context) -> String {
239        let mut key = String::with_capacity(64);
240        key.push_str(&self.id);
241        for component in &self.cache_key {
242            key.push('\u{1}');
243            key.push_str(&crate::vars::interpolate(ctx, component));
244        }
245        key
246    }
247}
248
249/// Reads a `Vec<u16>` of HTTP statuses from a JSON array, if present and valid.
250fn parse_statuses(v: Option<&serde_json::Value>) -> Result<Option<Vec<u16>>, String> {
251    let Some(v) = v else { return Ok(None) };
252    let arr = v
253        .as_array()
254        .ok_or("proxy-cache: cache_http_statuses must be an array of integers")?;
255    let mut out = Vec::with_capacity(arr.len());
256    for item in arr {
257        let n = item
258            .as_u64()
259            .ok_or("proxy-cache: cache_http_statuses entries must be integers")?;
260        if !(200..=599).contains(&n) {
261            return Err(format!(
262                "proxy-cache: cache status {} is out of range (200-599)",
263                n
264            ));
265        }
266        out.push(n as u16);
267    }
268    if out.is_empty() {
269        return Err("proxy-cache: cache_http_statuses must not be empty".to_string());
270    }
271    Ok(Some(out))
272}
273
274#[async_trait]
275impl Plugin for ProxyCachePlugin {
276    fn plugin_type(&self) -> &str {
277        "proxy-cache"
278    }
279
280    async fn execute(
281        &self,
282        mut ctx: Context,
283        _named_inputs: &HashMap<String, serde_json::Value>,
284    ) -> PluginResult {
285        // Non-cacheable methods bypass the cache entirely in both phases.
286        if !self.method_cacheable(&ctx) {
287            return Ok(PluginOutput {
288                context: ctx,
289                named_outputs: HashMap::new(),
290            });
291        }
292
293        let key = self.derive_key(&ctx);
294
295        match self.role {
296            Role::Lookup => {
297                if let Some(entry) = self.resources.traffic.cache.get(&key) {
298                    // Hit: serve the cached response and short-circuit to the
299                    // client via the error port (→ client.in).
300                    ctx.response.status_code = entry.status;
301                    ctx.response.headers = entry.headers;
302                    ctx.response.body = entry.body;
303                    if self.hide_cache_headers {
304                        for h in HIDDEN_HEADERS {
305                            ctx.response.headers.remove(*h);
306                        }
307                    }
308                    ctx.response
309                        .headers
310                        .insert(CACHE_STATUS_HEADER.to_string(), vec!["HIT".to_string()]);
311
312                    let error = GatewayError {
313                        node_id: String::new(),
314                        code: "PROXY_CACHE_HIT".to_string(),
315                        message: "Served from cache".to_string(),
316                        metadata: HashMap::new(),
317                    };
318                    return Err(PluginExecutionError {
319                        context: ctx,
320                        error,
321                    });
322                }
323                // Miss: continue to the upstream.
324                Ok(PluginOutput {
325                    context: ctx,
326                    named_outputs: HashMap::new(),
327                })
328            }
329            Role::Store => {
330                let status = ctx.response.status_code;
331                if self.cache_statuses.contains(&status) {
332                    self.resources.traffic.cache.put(
333                        key,
334                        status,
335                        ctx.response.headers.clone(),
336                        ctx.response.body.clone(),
337                        self.cache_ttl,
338                    );
339                }
340                // This response came from the upstream, not the cache.
341                ctx.response
342                    .headers
343                    .insert(CACHE_STATUS_HEADER.to_string(), vec!["MISS".to_string()]);
344                Ok(PluginOutput {
345                    context: ctx,
346                    named_outputs: HashMap::new(),
347                })
348            }
349        }
350    }
351}
352
353#[cfg(test)]
354mod tests {
355    use super::*;
356    use crate::context::{GatewayRequest, GatewayResponse, Protocol};
357    use bytes::Bytes;
358
359    fn ctx(method: &str) -> Context {
360        Context {
361            request: GatewayRequest {
362                method: method.to_string(),
363                path: "/products".to_string(),
364                host: "shop.example".to_string(),
365                scheme: "http".to_string(),
366                headers: HashMap::new(),
367                query_params: HashMap::new(),
368                body: Bytes::new(),
369                remote_addr: "10.0.0.1:5000".to_string(),
370                protocol: Protocol::Http1,
371            },
372            response: GatewayResponse {
373                status_code: 0,
374                headers: HashMap::new(),
375                body: Bytes::new(),
376            },
377            message: HashMap::new(),
378            errors: Vec::new(),
379        }
380    }
381
382    fn cfg(pairs: &[(&str, serde_json::Value)]) -> HashMap<String, serde_json::Value> {
383        pairs
384            .iter()
385            .map(|(k, v)| (k.to_string(), v.clone()))
386            .collect()
387    }
388
389    fn lookup(r: &Arc<PluginResources>) -> ProxyCachePlugin {
390        ProxyCachePlugin::from_config(
391            &cfg(&[
392                ("phase", serde_json::json!("lookup")),
393                ("id", serde_json::json!("cat")),
394            ]),
395            r,
396        )
397        .unwrap()
398    }
399
400    fn store(r: &Arc<PluginResources>) -> ProxyCachePlugin {
401        ProxyCachePlugin::from_config(
402            &cfg(&[
403                ("phase", serde_json::json!("store")),
404                ("id", serde_json::json!("cat")),
405            ]),
406            r,
407        )
408        .unwrap()
409    }
410
411    #[test]
412    fn test_missing_id_and_bad_role_fail() {
413        let r = PluginResources::empty();
414        assert!(
415            ProxyCachePlugin::from_config(&cfg(&[("phase", serde_json::json!("lookup"))]), &r)
416                .is_err()
417        );
418        assert!(ProxyCachePlugin::from_config(
419            &cfg(&[
420                ("phase", serde_json::json!("bogus")),
421                ("id", serde_json::json!("x"))
422            ]),
423            &r
424        )
425        .is_err());
426    }
427
428    #[test]
429    fn test_key_derivation_is_deterministic_and_shared() {
430        let r = PluginResources::empty();
431        let l = lookup(&r);
432        let s = store(&r);
433        // Both nodes derive the same key from the same request + config.
434        assert_eq!(l.derive_key(&ctx("GET")), s.derive_key(&ctx("GET")));
435        // Method participates in the default key.
436        assert_ne!(l.derive_key(&ctx("GET")), l.derive_key(&ctx("HEAD")));
437    }
438
439    #[tokio::test]
440    async fn test_store_then_lookup_returns_hit() {
441        let r = PluginResources::empty();
442        let l = lookup(&r);
443        let s = store(&r);
444
445        // Cold lookup → miss (passes through).
446        let miss = l.execute(ctx("GET"), &HashMap::new()).await;
447        assert!(miss.is_ok(), "cold lookup should miss and pass through");
448
449        // Upstream produced a 200 body → store caches it.
450        let mut resp = ctx("GET");
451        resp.response.status_code = 200;
452        resp.response.body = Bytes::from_static(b"cached-body");
453        let stored = s.execute(resp, &HashMap::new()).await.unwrap();
454        assert_eq!(
455            stored.context.response.headers.get(CACHE_STATUS_HEADER),
456            Some(&vec!["MISS".to_string()])
457        );
458
459        // Warm lookup → hit, short-circuits with the cached body.
460        let hit = l
461            .execute(ctx("GET"), &HashMap::new())
462            .await
463            .expect_err("warm lookup should hit and short-circuit");
464        assert_eq!(hit.error.code, "PROXY_CACHE_HIT");
465        assert_eq!(hit.context.response.status_code, 200);
466        assert_eq!(
467            hit.context.response.body,
468            Bytes::from_static(b"cached-body")
469        );
470        assert_eq!(
471            hit.context.response.headers.get(CACHE_STATUS_HEADER),
472            Some(&vec!["HIT".to_string()])
473        );
474    }
475
476    #[tokio::test]
477    async fn test_non_cacheable_method_passes_through() {
478        let r = PluginResources::empty();
479        let l = lookup(&r);
480        let s = store(&r);
481
482        // POST is not in the default cache_method → both phases pass through.
483        let mut resp = ctx("POST");
484        resp.response.status_code = 200;
485        resp.response.body = Bytes::from_static(b"not-cached");
486        s.execute(resp, &HashMap::new()).await.unwrap();
487
488        let out = l.execute(ctx("POST"), &HashMap::new()).await;
489        assert!(out.is_ok(), "non-cacheable method must never hit the cache");
490    }
491}
featherbit/plugins/native/proxy_cache.rs

featherbit/plugins/native/
proxy_cache.rs
