Bump from upstream

.bleep

@@ -1 +1 @@
-ed8657309187516d2e673037821a9fbd8405d703
+4842474141427ccc0d962b42e31c285b203da09d

.github/workflows/build.yml

@@ -8,7 +8,7 @@ jobs:
       fail-fast: false
       matrix:
         # nightly, msrv, and latest stable
-        toolchain: [nightly, 1.83.0, 1.87.0]
+        toolchain: [nightly, 1.84.0, 1.91.1]
     runs-on: ubuntu-latest
     # Only run on "pull_request" event for external PRs. This is to avoid
     # duplicate builds for PRs created from internal branches.
@@ -48,12 +48,12 @@ jobs:
 
       - name: Run cargo clippy
         run: |
-          [[ ${{ matrix.toolchain }} != 1.87.0 ]] || cargo clippy --all-targets --all -- --allow=unknown-lints --deny=warnings
+          [[ ${{ matrix.toolchain }} != 1.91.1 ]] || cargo clippy --all-targets --all -- --allow=unknown-lints --deny=warnings
 
       - name: Run cargo audit
         run: |
-          [[ ${{ matrix.toolchain }} != 1.87.0 ]] || (cargo install --locked cargo-audit && cargo audit)
+          [[ ${{ matrix.toolchain }} != 1.91.1 ]] || (cargo install --locked cargo-audit && cargo audit)
 
       - name: Run cargo machete
         run: |
-          [[ ${{ matrix.toolchain }} != 1.87.0 ]] || (cargo install cargo-machete --version 0.7.0 && cargo machete)
+          [[ ${{ matrix.toolchain }} != 1.91.1 ]] || (cargo install cargo-machete --version 0.7.0 && cargo machete)

Cargo.toml

@@ -29,12 +29,14 @@ members = [
 ]
 
 [workspace.dependencies]
+bstr = "1.12.0"
 tokio = "1"
+tokio-stream = { version = "0.1" }
 async-trait = "0.1.42"
 httparse = "1"
 bytes = "1.0"
 derivative = "2.2.0"
-http = "1.0.0"
+http = "1"
 log = "0.4"
 h2 = ">=0.4.11"
 once_cell = "1"

README.md

@@ -61,9 +61,9 @@ Both x86_64 and aarch64 architectures will be supported.
 
 Pingora keeps a rolling MSRV (minimum supported Rust version) policy of 6 months. This means we will accept PRs that upgrade the MSRV as long as the new Rust version used is at least 6 months old.
 
-Our current MSRV is effectively 1.83.
+Our current MSRV is 1.84.
 
-Previously Pingora advertised an MSRV of 1.72. Older Rust versions may still be able to compile via `cargo update` pinning dependencies such as `backtrace@0.3.74`. The advertised MSRV in config files will be officially bumped to 1.83 in an upcoming release.
+Building with the optional feature `boringssl` with Boring >= 4.14 requires Rust 1.80.
 
 ## Build Requirements
 

clippy.toml

@@ -1 +1 @@
-msrv = "1.72"
+msrv = "1.84"

pingora-boringssl/src/boring_tokio.rs

@@ -263,9 +263,7 @@ where
                 return Poll::Pending;
             }
             Err(e) => {
-                return Poll::Ready(Err(e
-                    .into_io_error()
-                    .unwrap_or_else(|e| io::Error::new(io::ErrorKind::Other, e))));
+                return Poll::Ready(Err(e.into_io_error().unwrap_or_else(io::Error::other)));
             }
         }
 

pingora-cache/Cargo.toml

@@ -23,6 +23,7 @@ pingora-header-serde = { version = "0.6.0", path = "../pingora-header-serde" }
 pingora-http = { version = "0.6.0", path = "../pingora-http" }
 pingora-lru = { version = "0.6.0", path = "../pingora-lru" }
 pingora-timeout = { version = "0.6.0", path = "../pingora-timeout" }
+bstr = { workspace = true }
 http = { workspace = true }
 indexmap = "1"
 once_cell = { workspace = true }

pingora-cache/src/cache_control.rs

@@ -255,13 +255,13 @@ impl CacheControl {
         self.has_key_without_value("private")
     }
 
-    fn get_field_names(&self, key: &str) -> Option<ListValueIter> {
+    fn get_field_names(&self, key: &str) -> Option<ListValueIter<'_>> {
         let value = self.directives.get(key)?.as_ref()?;
         Some(ListValueIter::from(value))
     }
 
     /// Get the values of `private=`
-    pub fn private_field_names(&self) -> Option<ListValueIter> {
+    pub fn private_field_names(&self) -> Option<ListValueIter<'_>> {
         self.get_field_names("private")
     }
 
@@ -271,7 +271,7 @@ impl CacheControl {
     }
 
     /// Get the values of `no-cache=`
-    pub fn no_cache_field_names(&self) -> Option<ListValueIter> {
+    pub fn no_cache_field_names(&self) -> Option<ListValueIter<'_>> {
         self.get_field_names("no-cache")
     }
 

pingora-cache/src/eviction/lru.rs

@@ -62,6 +62,29 @@ impl<const N: usize> Manager<N> {
         Manager(Lru::with_capacity_and_watermark(limit, capacity, watermark))
     }
 
+    /// Get the number of shards
+    pub fn shards(&self) -> usize {
+        self.0.shards()
+    }
+
+    /// Get the weight (total size) of a specific shard
+    pub fn shard_weight(&self, shard: usize) -> usize {
+        self.0.shard_weight(shard)
+    }
+
+    /// Get the number of items in a specific shard
+    pub fn shard_len(&self, shard: usize) -> usize {
+        self.0.shard_len(shard)
+    }
+
+    /// Get the shard index for a given cache key
+    ///
+    /// This allows callers to know which shard was affected by an operation
+    /// without acquiring any locks.
+    pub fn get_shard_for_key(&self, key: &CompactCacheKey) -> usize {
+        (u64key(key) % N as u64) as usize
+    }
+
     /// Serialize the given shard
     pub fn serialize_shard(&self, shard: usize) -> Result<Vec<u8>> {
         use rmp_serde::encode::Serializer;
@@ -101,6 +124,12 @@ impl<const N: usize> Manager<N> {
             .or_err(InternalError, "when deserializing LRU")?;
         Ok(())
     }
+
+    /// Peek the weight associated with a cache key without changing its LRU order.
+    pub fn peek_weight(&self, item: &CompactCacheKey) -> Option<usize> {
+        let key = u64key(item);
+        self.0.peek_weight(key)
+    }
 }
 
 struct InsertToManager<'a, const N: usize> {
@@ -171,9 +200,14 @@ impl<const N: usize> EvictionManager for Manager<N> {
             .collect()
     }
 
-    fn increment_weight(&self, item: CompactCacheKey, delta: usize) -> Vec<CompactCacheKey> {
-        let key = u64key(&item);
-        self.0.increment_weight(key, delta);
+    fn increment_weight(
+        &self,
+        item: &CompactCacheKey,
+        delta: usize,
+        max_weight: Option<usize>,
+    ) -> Vec<CompactCacheKey> {
+        let key = u64key(item);
+        self.0.increment_weight(key, delta, max_weight);
         self.0
             .evict_to_limit()
             .into_iter()

pingora-cache/src/eviction/mod.rs

@@ -60,10 +60,18 @@ pub trait EvictionManager: Send + Sync {
     /// Adjust an item's weight upwards by a delta. If the item is not already admitted,
     /// nothing will happen.
     ///
+    /// An optional `max_weight` hint indicates the known max weight of the current key in case the
+    /// weight should not be incremented above this amount.
+    ///
     /// Return one or more items to evict. The sizes of these items are deducted
     /// from the total size already. The caller needs to make sure that these assets are actually
     /// removed from the storage.
-    fn increment_weight(&self, item: CompactCacheKey, delta: usize) -> Vec<CompactCacheKey>;
+    fn increment_weight(
+        &self,
+        item: &CompactCacheKey,
+        delta: usize,
+        max_weight: Option<usize>,
+    ) -> Vec<CompactCacheKey>;
 
     /// Remove an item from the eviction manager.
     ///

pingora-cache/src/eviction/simple_lru.rs

@@ -124,7 +124,7 @@ impl Manager {
         if self.used.load(Ordering::Relaxed) <= self.limit
             && self
                 .items_watermark
-                .map_or(true, |w| self.items.load(Ordering::Relaxed) <= w)
+                .is_none_or(|w| self.items.load(Ordering::Relaxed) <= w)
         {
             return vec![];
         }
@@ -235,8 +235,13 @@ impl EvictionManager for Manager {
         self.evict()
     }
 
-    fn increment_weight(&self, item: CompactCacheKey, delta: usize) -> Vec<CompactCacheKey> {
-        let key = u64key(&item);
+    fn increment_weight(
+        &self,
+        item: &CompactCacheKey,
+        delta: usize,
+        _max_weight: Option<usize>,
+    ) -> Vec<CompactCacheKey> {
+        let key = u64key(item);
         self.increase_weight(key, delta);
         self.evict()
     }

pingora-cache/src/filters.rs

@@ -89,7 +89,7 @@ pub fn calculate_fresh_until(
     if authorization_present {
         let uncacheable = cache_control
             .as_ref()
-            .map_or(true, |cc| !cc.allow_caching_authorized_req());
+            .is_none_or(|cc| !cc.allow_caching_authorized_req());
         if uncacheable {
             return None;
         }

pingora-cache/src/hashtable.rs

@@ -49,11 +49,11 @@ where
     }
 
     #[allow(dead_code)]
-    pub fn read(&self, key: u128) -> RwLockReadGuard<HashMap<u128, V>> {
+    pub fn read(&self, key: u128) -> RwLockReadGuard<'_, HashMap<u128, V>> {
         self.get(key).read()
     }
 
-    pub fn write(&self, key: u128) -> RwLockWriteGuard<HashMap<u128, V>> {
+    pub fn write(&self, key: u128) -> RwLockWriteGuard<'_, HashMap<u128, V>> {
         self.get(key).write()
     }
 
@@ -103,7 +103,7 @@ where
     pub fn new(shard_capacity: usize) -> Self {
         use std::num::NonZeroUsize;
         // safe, 1 != 0
-        const ONE: NonZeroUsize = unsafe { NonZeroUsize::new_unchecked(1) };
+        const ONE: NonZeroUsize = NonZeroUsize::new(1).unwrap();
         let mut cache = ConcurrentLruCache {
             lrus: Default::default(),
         };
@@ -119,11 +119,11 @@ where
     }
 
     #[allow(dead_code)]
-    pub fn read(&self, key: u128) -> RwLockReadGuard<LruCache<u128, V>> {
+    pub fn read(&self, key: u128) -> RwLockReadGuard<'_, LruCache<u128, V>> {
         self.get(key).read()
     }
 
-    pub fn write(&self, key: u128) -> RwLockWriteGuard<LruCache<u128, V>> {
+    pub fn write(&self, key: u128) -> RwLockWriteGuard<'_, LruCache<u128, V>> {
         self.get(key).write()
     }
 

pingora-cache/src/lib.rs

@@ -688,7 +688,7 @@ impl HttpCache {
         self.inner_mut()
             .max_file_size_tracker
             .as_mut()
-            .map_or(true, |t| t.add_body_bytes(bytes_len))
+            .is_none_or(|t| t.add_body_bytes(bytes_len))
     }
 
     /// Check if the max file size has been exceeded according to max file size tracker.
@@ -823,6 +823,18 @@ impl HttpCache {
         }
     }
 
+    /// Return whether the underlying storage backend supports streaming partial write.
+    ///
+    /// Returns None if cache is not enabled.
+    pub fn support_streaming_partial_write(&self) -> Option<bool> {
+        self.inner.as_ref().and_then(|inner| {
+            inner
+                .enabled_ctx
+                .as_ref()
+                .map(|c| c.storage.support_streaming_partial_write())
+        })
+    }
+
     /// Call this when cache hit is fully read.
     ///
     /// This call will release resource if any and log the timing in tracing if set.
@@ -969,8 +981,8 @@ impl HttpCache {
                         MissFinishType::Created(size) => {
                             eviction.admit(cache_key, size, meta.0.internal.fresh_until)
                         }
-                        MissFinishType::Appended(size) => {
-                            eviction.increment_weight(cache_key, size)
+                        MissFinishType::Appended(size, max_size) => {
+                            eviction.increment_weight(&cache_key, size, max_size)
                         }
                     };
                     // actual eviction can be done async
@@ -1250,6 +1262,18 @@ impl HttpCache {
         }
     }
 
+    /// Return the [`CacheKey`] of this asset if any.
+    ///
+    /// This is allowed to be called in any phase. If the cache key callback was not called,
+    /// this will return None.
+    pub fn maybe_cache_key(&self) -> Option<&CacheKey> {
+        (!matches!(
+            self.phase(),
+            CachePhase::Disabled(NoCacheReason::NeverEnabled) | CachePhase::Uninit
+        ))
+        .then(|| self.cache_key())
+    }
+
     /// Perform the cache lookup from the given cache storage with the given cache key
     ///
     /// A cache hit will return [CacheMeta] which contains the header and meta info about
@@ -1426,7 +1450,7 @@ impl HttpCache {
         let mut span = inner_enabled.traces.child("cache_lock");
         // should always call is_cache_locked() before this function, which should guarantee that
         // the inner cache has a read lock and lock ctx
-        if let Some(lock_ctx) = inner_enabled.lock_ctx.as_mut() {
+        let (read_lock, status) = if let Some(lock_ctx) = inner_enabled.lock_ctx.as_mut() {
             let lock = lock_ctx.lock.take(); // remove the lock from self
             if let Some(Locked::Read(r)) = lock {
                 let now = Instant::now();
@@ -1437,23 +1461,26 @@ impl HttpCache {
                         wait_timeout.saturating_sub(self.lock_duration().unwrap_or(Duration::ZERO));
                     match timeout(wait_timeout, r.wait()).await {
                         Ok(()) => r.lock_status(),
-                        // TODO: need to differentiate WaitTimeout vs. Lock(Age)Timeout (expired)?
-                        Err(_) => LockStatus::Timeout,
+                        Err(_) => LockStatus::WaitTimeout,
                     }
                 } else {
                     r.wait().await;
                     r.lock_status()
                 };
                 self.digest.add_lock_duration(now.elapsed());
-                let tag_value: &'static str = status.into();
-                span.set_tag(|| Tag::new("status", tag_value));
-                status
+                (r, status)
             } else {
                 panic!("cache_lock_wait on wrong type of lock")
             }
         } else {
             panic!("cache_lock_wait without cache lock")
+        };
+        if let Some(lock_ctx) = self.inner_enabled().lock_ctx.as_ref() {
+            lock_ctx
+                .cache_lock
+                .trace_lock_wait(&mut span, &read_lock, status);
         }
+        status
     }
 
     /// How long did this request wait behind the read lock