ext/cache/lscache.rs - main - githubmirror/deno

ext/cache/lscache.rs
416 строк11 KB
Nathan Whitaker
fix(cache): combine duplicate headers for Vary matching (#36476)
4 часа назад
4 часа назад74ddcdd
// Copyright 2018-2026 the Deno authors. MIT license.

use std::cell::RefCell;
use std::rc::Rc;

use async_stream::try_stream;
use base64::Engine;
use bytes::Bytes;
use deno_core::BufMutView;
use deno_core::Resource;
use deno_core::convert::ByteString;
use deno_core::unsync::spawn;
use futures::StreamExt;
use futures::TryStreamExt;
use http::HeaderMap;
use http::HeaderName;
use http::HeaderValue;
use http::header::VARY;
use http_body_util::combinators::UnsyncBoxBody;
use slab::Slab;

use crate::CacheDeleteRequest;
use crate::CacheError;
use crate::CacheKeyEntry;
use crate::CacheMatchRequest;
use crate::CacheMatchResponseMeta;
use crate::CachePutRequest;
use crate::CacheResponseResource;
use crate::get_header;
use crate::get_headers_from_vary_header;
use crate::lsc_shard::CacheShard;

const REQHDR_PREFIX: &str = "x-lsc-meta-reqhdr-";

#[derive(Clone, Default)]
pub struct LscBackend {
  shard: Rc<RefCell<Option<Rc<CacheShard>>>>,
  id2name: Rc<RefCell<Slab<String>>>,
}

impl LscBackend {
  pub fn set_shard(&self, shard: Rc<CacheShard>) {
    *self.shard.borrow_mut() = Some(shard);
  }
}

#[allow(clippy::unused_async, reason = "trait requires async interface")]
impl LscBackend {
  /// Open a cache storage. Internally, this allocates an id and maps it
  /// to the provided cache name.
  pub async fn storage_open(
    &self,
    cache_name: String,
  ) -> Result<i64, CacheError> {
    if cache_name.is_empty() {
      return Err(CacheError::EmptyName);
    }
    let id = self.id2name.borrow_mut().insert(cache_name);
    Ok(id as i64)
  }

  /// Check if a cache with the provided name exists. Always returns `true`.
  pub async fn storage_has(
    &self,
    _cache_name: String,
  ) -> Result<bool, CacheError> {
    Ok(true)
  }

  /// Delete a cache storage. Not yet implemented.
  pub async fn storage_delete(
    &self,
    _cache_name: String,
  ) -> Result<bool, CacheError> {
    Err(CacheError::DeletionNotSupported)
  }

  /// List all cache names currently known to this backend.
  pub async fn storage_keys(&self) -> Result<Vec<String>, CacheError> {
    let mut seen = std::collections::HashSet::new();
    let mut names = Vec::new();
    for (_, name) in self.id2name.borrow().iter() {
      if seen.insert(name.clone()) {
        names.push(name.clone());
      }
    }
    Ok(names)
  }

  /// Writes an entry to the cache.
  pub async fn put(
    &self,
    request_response: CachePutRequest,
    resource: Option<Rc<dyn Resource>>,
  ) -> Result<(), CacheError> {
    let Some(shard) = self.shard.borrow().as_ref().cloned() else {
      return Err(CacheError::NotAvailable);
    };

    let Some(cache_name) = self
      .id2name
      .borrow_mut()
      .get(request_response.cache_id as usize)
      .cloned()
    else {
      return Err(CacheError::NotFound);
    };
    let object_key = build_cache_object_key(
      cache_name.as_bytes(),
      request_response.request_url.as_bytes(),
    );
    let mut headers = HeaderMap::new();
    for hdr in &request_response.request_headers {
      headers.append(
        HeaderName::from_bytes(
          &[REQHDR_PREFIX.as_bytes(), &hdr.0[..]].concat(),
        )?,
        HeaderValue::from_bytes(&hdr.1[..])?,
      );
    }
    for hdr in &request_response.response_headers {
      if hdr.0.starts_with(b"x-lsc-meta-") {
        continue;
      }
      if hdr.0[..] == b"content-encoding"[..] {
        return Err(CacheError::ContentEncodingNotAllowed);
      }
      headers.append(
        HeaderName::from_bytes(&hdr.0[..])?,
        HeaderValue::from_bytes(&hdr.1[..])?,
      );
    }

    headers.insert(
      HeaderName::from_bytes(b"x-lsc-meta-cached-at")?,
      HeaderValue::from_bytes(
        chrono::Utc::now()
          .to_rfc3339_opts(chrono::SecondsFormat::Secs, true)
          .as_bytes(),
      )?,
    );

    let body = try_stream! {
      if let Some(resource) = resource {
        loop {
          let (size, buf) = resource.clone().read_byob(BufMutView::new(64 * 1024)).await.map_err(CacheError::Other)?;
          if size == 0 {
            break;
          }
          yield Bytes::copy_from_slice(&buf[..size]);
        }
      }
    };
    let (body_tx, body_rx) = futures::channel::mpsc::channel(4);
    spawn(body.map(Ok::<Result<_, CacheError>, _>).forward(body_tx));
    let body = http_body_util::StreamBody::new(
      body_rx.into_stream().map_ok(http_body::Frame::data),
    );
    let body = UnsyncBoxBody::new(body);
    shard.put_object(&object_key, headers, body).await?;
    Ok(())
  }

  /// Matches a request against the cache.
  pub async fn r#match(
    &self,
    request: CacheMatchRequest,
  ) -> Result<
    Option<(CacheMatchResponseMeta, Option<CacheResponseResource>)>,
    CacheError,
  > {
    let Some(shard) = self.shard.borrow().as_ref().cloned() else {
      return Err(CacheError::NotAvailable);
    };
    let Some(cache_name) = self
      .id2name
      .borrow()
      .get(request.cache_id as usize)
      .cloned()
    else {
      return Err(CacheError::NotFound);
    };
    let object_key = build_cache_object_key(
      cache_name.as_bytes(),
      request.request_url.as_bytes(),
    );
    let Some(res) = shard.get_object(&object_key).await? else {
      return Ok(None);
    };

    // Is this a tombstone?
    if res.headers().contains_key("x-lsc-meta-deleted-at") {
      return Ok(None);
    }

    // From https://w3c.github.io/ServiceWorker/#request-matches-cached-item-algorithm
    // If there's Vary header in the response, ensure all the
    // headers of the cached request match the query request.
    if let Some(vary_header) = get_header_from_map(VARY.as_str(), res.headers())
      && !vary_header_matches(
        &vary_header,
        &request.request_headers,
        res.headers(),
      )
    {
      return Ok(None);
    }

    let mut response_headers: Vec<(ByteString, ByteString)> = res
      .headers()
      .iter()
      .filter_map(|(k, v)| {
        if k.as_str().starts_with("x-lsc-meta-") || k.as_str() == "x-ryw" {
          None
        } else {
          Some((k.as_str().into(), v.as_bytes().into()))
        }
      })
      .collect();

    if let Some(x) = res
      .headers()
      .get("x-lsc-meta-cached-at")
      .and_then(|x| x.to_str().ok())
      && let Ok(cached_at) = chrono::DateTime::parse_from_rfc3339(x)
    {
      let age = chrono::Utc::now()
        .signed_duration_since(cached_at)
        .num_seconds();
      if age >= 0 {
        response_headers.push(("age".into(), age.to_string().into()));
      }
    }

    let meta = CacheMatchResponseMeta {
      response_status: res.status().as_u16(),
      response_status_text: res
        .status()
        .canonical_reason()
        .unwrap_or("")
        .to_string(),
      request_headers: res
        .headers()
        .iter()
        .filter_map(|(k, v)| {
          let reqhdr_prefix = REQHDR_PREFIX.as_bytes();
          if k.as_str().as_bytes().starts_with(reqhdr_prefix) {
            Some((
              k.as_str().as_bytes()[REQHDR_PREFIX.len()..].into(),
              v.as_bytes().into(),
            ))
          } else {
            None
          }
        })
        .collect(),
      response_headers,
    };

    let body = http_body_util::BodyDataStream::new(res.into_body())
      .into_stream()
      .map_err(std::io::Error::other);
    let body = CacheResponseResource::lsc(body);

    Ok(Some((meta, Some(body))))
  }

  pub async fn delete(
    &self,
    request: CacheDeleteRequest,
  ) -> Result<bool, CacheError> {
    let Some(shard) = self.shard.borrow().as_ref().cloned() else {
      return Err(CacheError::NotAvailable);
    };

    let Some(cache_name) = self
      .id2name
      .borrow_mut()
      .get(request.cache_id as usize)
      .cloned()
    else {
      return Err(CacheError::NotFound);
    };
    let object_key = build_cache_object_key(
      cache_name.as_bytes(),
      request.request_url.as_bytes(),
    );
    let mut headers = HeaderMap::new();
    headers.insert(
      HeaderName::from_bytes(b"expires")?,
      HeaderValue::from_bytes(b"Thu, 01 Jan 1970 00:00:00 GMT")?,
    );
    headers.insert(
      HeaderName::from_bytes(b"x-lsc-meta-deleted-at")?,
      HeaderValue::from_bytes(
        chrono::Utc::now()
          .to_rfc3339_opts(chrono::SecondsFormat::Secs, true)
          .as_bytes(),
      )?,
    );
    shard.put_object_empty(&object_key, headers).await?;
    Ok(true)
  }

  /// Listing cached request keys is not supported by the remote cache
  /// backend as it provides no way to enumerate stored objects.
  pub async fn keys(
    &self,
    _cache_id: i64,
    _request_url: Option<String>,
  ) -> Result<Vec<CacheKeyEntry>, CacheError> {
    Err(CacheError::KeysNotSupported)
  }
}
impl deno_core::Resource for LscBackend {
  fn name(&self) -> std::borrow::Cow<'_, str> {
    "LscBackend".into()
  }
}

fn vary_header_matches(
  vary_header: &ByteString,
  query_request_headers: &[(ByteString, ByteString)],
  cached_headers: &HeaderMap,
) -> bool {
  let vary_header = match std::str::from_utf8(vary_header) {
    Ok(vary_header) => vary_header,
    Err(_) => return false,
  };
  let headers = get_headers_from_vary_header(vary_header);
  for header in headers {
    if header == "*" {
      return false;
    }
    // Ignoring `accept-encoding` is safe because we refuse to cache responses
    // with `content-encoding`
    if header == "accept-encoding" {
      continue;
    }
    let lookup_key = format!("{}{}", REQHDR_PREFIX, header);
    let query_header = get_header(&header, query_request_headers);
    let cached_header = get_header_from_map(&lookup_key, cached_headers);
    if query_header != cached_header {
      return false;
    }
  }
  true
}

fn get_header_from_map(name: &str, headers: &HeaderMap) -> Option<ByteString> {
  let mut values = headers.get_all(name).iter();
  let mut combined = values.next()?.as_bytes().to_vec();
  for value in values {
    combined.extend_from_slice(b", ");
    combined.extend_from_slice(value.as_bytes());
  }
  Some(combined.into())
}

fn build_cache_object_key(cache_name: &[u8], request_url: &[u8]) -> String {
  format!(
    "v1/{}/{}",
    base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(cache_name),
    base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(request_url),
  )
}

#[cfg(test)]
mod tests {
  use super::*;

  #[test]
  fn lsc_vary_header_matches_combined_values() {
    let mut cached_headers = HeaderMap::new();
    cached_headers.append(
      "x-lsc-meta-reqhdr-x-example",
      HeaderValue::from_static("first"),
    );
    cached_headers.append(
      "x-lsc-meta-reqhdr-x-example",
      HeaderValue::from_static("cached"),
    );

    let query_headers = vec![
      ("X-Example".into(), "first".into()),
      ("x-example".into(), "cached".into()),
    ];
    assert!(vary_header_matches(
      &"x-example".into(),
      &query_headers,
      &cached_headers,
    ));

    let query_headers = vec![
      ("x-example".into(), "first".into()),
      ("x-example".into(), "query".into()),
    ];
    assert!(!vary_header_matches(
      &"x-example".into(),
      &query_headers,
      &cached_headers,
    ));
  }

  #[test]
  fn lsc_combines_repeated_vary_fields() {
    let mut headers = HeaderMap::new();
    headers.append(VARY, HeaderValue::from_static("accept-language"));
    headers.append(VARY, HeaderValue::from_static("x-example"));

    assert_eq!(
      get_header_from_map(VARY.as_str(), &headers),
      Some("accept-language, x-example".into()),
    );
  }
}