Repository: cloudflare/pingora (language: rust)

Issue:
`Default` trait implementation for `Opt` shouldn’t have side-effects
## Describe the bug

As things are right now, `Opt::default()` on `pingora_core::server::configuration::Opt` is implemented as an alias for `Opt::parse()`:

https://github.com/cloudflare/pingora/blob/216d8e9d92c75d46cba78da9b167fb41dec91ad7/pingora-core/src/server/configuration/mod.rs#L168-L172

This can cause unexpected side-effects if someone (like me) gets some Pingora options by other means and merely wants to fill up the data. Depending on the command line parameters the application might even shut down.

## Pingora info

**Pingora version**: 0.2.0 or [216d8e9](https://github.com/cloudflare/pingora/commit/216d8e9d92c75d46cba78da9b167fb41dec91ad7)

## Steps to reproduce

Consider the following application:

```rust
use pingora_core::server::{configuration::{Opt, ServerConf}, Server};

fn main() {
    let server = Server::new_with_opt_and_conf(
        Opt {
            daemon: true,
            ..Default::default()
        },
        ServerConf::default(),
    );
    println!("Server created");
}
```

Run it using `cargo run -- -h` or `cargo run -- --blub` command.

## Expected results

The application should be able to create a server regardless of its command line flags and print: “Server created”

## Observed results

Using the first command, the application prints command line help which doesn’t apply to the application and exits. Using the first command, it produces the error message “error: Found argument '--blub' which wasn't expected, or isn't valid in this context” and exits. In both cases the message is never printed.

## Additional context

The `Default` trait implementation can be derived automatically here, it will set all the boolean flags to `false` and the configuration path to `None`. If a trait-less alias for `Opt::parse()` is required, it can be provided under a dedicated and unambiguous name like `Opt::parse_command_line()`.


Discussion / hints:

Fix patch:
diff --git a/.bleep b/.bleep
index 84aa6357c..cee92a797 100644
--- a/.bleep
+++ b/.bleep
@@ -1,1 +1,1 @@
-f70d8b77a4085cbe11b9559317f6d6e7e49914db
\ No newline at end of file
+361d88592075f7f98f581b139d0349f1b70190a2
\ No newline at end of file
diff --git a/docs/quick_start.md b/docs/quick_start.md
index 5099f7ec0..8a3b796b4 100644
--- a/docs/quick_start.md
+++ b/docs/quick_start.md
@@ -237,7 +237,7 @@ take advantage of with single-line change.
 
 ```rust
 fn main() {
-    let mut my_server = Server::new(Some(Opt::default())).unwrap();
+    let mut my_server = Server::new(Some(Opt::parse_args())).unwrap();
     ...
 }
 ```
diff --git a/pingora-cache/src/lib.rs b/pingora-cache/src/lib.rs
index 563727cfd..6ba057e30 100644
--- a/pingora-cache/src/lib.rs
+++ b/pingora-cache/src/lib.rs
@@ -44,7 +44,7 @@ pub use key::CacheKey;
 use lock::{CacheLock, LockStatus, Locked};
 pub use memory::MemCache;
 pub use meta::{CacheMeta, CacheMetaDefaults};
-pub use storage::{HitHandler, MissHandler, Storage};
+pub use storage::{HitHandler, MissHandler, PurgeType, Storage};
 pub use variance::VarianceBuilder;
 
 pub mod prelude {}
@@ -658,7 +658,10 @@ impl HttpCache {
                     let handle = span.handle();
                     for item in evicted {
                         // TODO: warn/log the error
-                        let _ = inner.storage.purge(&item, &handle).await;
+                        let _ = inner
+                            .storage
+                            .purge(&item, PurgeType::Eviction, &handle)
+                            .await;
                     }
                 }
                 inner.traces.finish_miss_span();
@@ -1063,7 +1066,10 @@ impl HttpCache {
                 let inner = self.inner_mut();
                 let mut span = inner.traces.child("purge");
                 let key = inner.key.as_ref().unwrap().to_compact();
-                let result = inner.storage.purge(&key, &span.handle()).await;
+                let result = inner
+                    .storage
+                    .purge(&key, PurgeType::Invalidation, &span.handle())
+                    .await;
                 // FIXME: also need to remove from eviction manager
                 span.set_tag(|| trace::Tag::new("purged", matches!(result, Ok(true))));
                 result
diff --git a/pingora-cache/src/memory.rs b/pingora-cache/src/memory.rs
index 3863b2bd3..dec8f1ad3 100644
--- a/pingora-cache/src/memory.rs
+++ b/pingora-cache/src/memory.rs
@@ -306,7 +306,12 @@ impl Storage for MemCache {
         Ok(Box::new(miss_handler))
     }
 
-    async fn purge(&'static self, key: &CompactCacheKey, _trace: &SpanHandle) -> Result<bool> {
+    async fn purge(
+        &'static self,
+        key: &CompactCacheKey,
+        _type: PurgeType,
+        _trace: &SpanHandle,
+    ) -> Result<bool> {
         // This usually purges the primary key because, without a lookup, the variance key is usually
         // empty
         let hash = key.combined();
@@ -525,7 +530,9 @@ mod test {
 
         assert!(cache.temp.read().contains_key(&hash));
 
-        let result = cache.purge(&key, &Span::inactive().handle()).await;
+        let result = cache
+            .purge(&key, PurgeType::Invalidation, &Span::inactive().handle())
+            .await;
         assert!(result.is_ok());
 
         assert!(!cache.temp.read().contains_key(&hash));
@@ -551,7 +558,9 @@ mod test {
 
         assert!(cache.cached.read().contains_key(&hash));
 
-        let result = cache.purge(&key, &Span::inactive().handle()).await;
+        let result = cache
+            .purge(&key, PurgeType::Invalidation, &Span::inactive().handle())
+            .await;
         assert!(result.is_ok());
 
         assert!(!cache.cached.read().contains_key(&hash));
diff --git a/pingora-cache/src/put.rs b/pingora-cache/src/put.rs
index e29cfadfb..e1041e00f 100644
--- a/pingora-cache/src/put.rs
+++ b/pingora-cache/src/put.rs
@@ -119,7 +119,7 @@ impl<C: CachePut> CachePutCtx<C> {
                 .handle();
             for item in evicted {
                 // TODO: warn/log the error
-                let _ = self.storage.purge(&item, &trace).await;
+                let _ = self.storage.purge(&item, PurgeType::Eviction, &trace).await;
             }
         }
 
diff --git a/pingora-cache/src/storage.rs b/pingora-cache/src/storage.rs
index c6365c7cb..0d4e1041b 100644
--- a/pingora-cache/src/storage.rs
+++ b/pingora-cache/src/storage.rs
@@ -22,6 +22,15 @@ use async_trait::async_trait;
 use pingora_error::Result;
 use std::any::Any;
 
+/// The reason a purge() is called
+#[derive(Debug, Clone, Copy)]
+pub enum PurgeType {
+    // For eviction because the cache storage is full
+    Eviction,
+    // For cache invalidation
+    Invalidation,
+}
+
 /// Cache storage interface
 #[async_trait]
 pub trait Storage {
@@ -45,7 +54,12 @@ pub trait Storage {
     /// Delete the cached asset for the given key
     ///
     /// [CompactCacheKey] is used here because it is how eviction managers store the keys
-    async fn purge(&'static self, key: &CompactCacheKey, trace: &SpanHandle) -> Result<bool>;
+    async fn purge(
+        &'static self,
+        key: &CompactCacheKey,
+        purge_type: PurgeType,
+        trace: &SpanHandle,
+    ) -> Result<bool>;
 
     /// Update cache header and metadata for the already stored asset.
     async fn update_meta(
diff --git a/pingora-core/src/connectors/http/v1.rs b/pingora-core/src/connectors/http/v1.rs
index 7958a0939..66f7bbba7 100644
--- a/pingora-core/src/connectors/http/v1.rs
+++ b/pingora-core/src/connectors/http/v1.rs
@@ -51,10 +51,11 @@ impl Connector {
 
     pub async fn release_http_session<P: Peer + Send + Sync + 'static>(
         &self,
-        session: HttpSession,
+        mut session: HttpSession,
         peer: &P,
         idle_timeout: Option<Duration>,
     ) {
+        session.respect_keepalive();
         if let Some(stream) = session.reuse().await {
             self.transport
                 .release_stream(stream, peer.reuse_hash(), idle_timeout);
diff --git a/pingora-core/src/connectors/http/v2.rs b/pingora-core/src/connectors/http/v2.rs
index 2c11ede17..a0829431a 100644
--- a/pingora-core/src/connectors/http/v2.rs
+++ b/pingora-core/src/connectors/http/v2.rs
@@ -102,6 +102,10 @@ impl ConnectionRef {
         &self.0.digest
     }
 
+    pub fn digest_mut(&mut self) -> Option<&mut Digest> {
+        Arc::get_mut(&mut self.0).map(|inner| &mut inner.digest)
+    }
+
     pub fn ping_timedout(&self) -> bool {
         self.0.ping_timeout_occurred.load(Ordering::Relaxed)
     }
diff --git a/pingora-core/src/listeners/tls.rs b/pingora-core/src/listeners/tls.rs
index 655346fd8..1dd63d462 100644
--- a/pingora-core/src/listeners/tls.rs
+++ b/pingora-core/src/listeners/tls.rs
@@ -38,6 +38,15 @@ pub struct TlsSettings {
     callbacks: Option<TlsAcceptCallbacks>,
 }
 
+impl From<SslAcceptorBuilder> for TlsSettings {
+    fn from(settings: SslAcceptorBuilder) -> Self {
+        TlsSettings {
+            accept_builder: settings,
+            callbacks: None,
+        }
+    }
+}
+
 impl Deref for TlsSettings {
     type Target = SslAcceptorBuilder;
 
diff --git a/pingora-core/src/modules/http/mod.rs b/pingora-core/src/modules/http/mod.rs
index 31c5f9a21..02a3e0c39 100644
--- a/pingora-core/src/modules/http/mod.rs
+++ b/pingora-core/src/modules/http/mod.rs
@@ -65,7 +65,7 @@ pub trait HttpModule {
     fn as_any_mut(&mut self) -> &mut dyn Any;
 }
 
-type Module = Box<dyn HttpModule + 'static + Send + Sync>;
+pub type Module = Box<dyn HttpModule + 'static + Send + Sync>;
 
 /// Trait to init the http module ctx for each request
 pub trait HttpModuleBuilder {
diff --git a/pingora-core/src/protocols/http/client.rs b/pingora-core/src/protocols/http/client.rs
index 5c44b6f2d..4d3044dcd 100644
--- a/pingora-core/src/protocols/http/client.rs
+++ b/pingora-core/src/protocols/http/client.rs
@@ -89,7 +89,9 @@ impl HttpSession {
 
     /// Set the write timeout for writing header and body.
     ///
-    /// The timeout is per write operation, not on the overall time writing the entire request
+    /// The timeout is per write operation, not on the overall time writing the entire request.
+    ///
+    /// This is a noop for h2.
     pub fn set_write_timeout(&mut self, timeout: Duration) {
         match self {
             HttpSession::H1(h1) => h1.write_timeout = Some(timeout),
@@ -151,7 +153,7 @@ impl HttpSession {
     /// Return the [Digest] of the connection
     ///
     /// For reused connection, the timing in the digest will reflect its initial handshakes
-    /// The caller should check if the connection is reused to avoid misuse of the timing field
+    /// The caller should check if the connection is reused to avoid misuse of the timing field.
     pub fn digest(&self) -> Option<&Digest> {
         match self {
             Self::H1(s) => Some(s.digest()),
@@ -159,6 +161,16 @@ impl HttpSession {
         }
     }
 
+    /// Return a mutable [Digest] reference for the connection, see [`digest`] for more details.
+    ///
+    /// Will return `None` if this is an H2 session and multiple streams are open.
+    pub fn digest_mut(&mut self) -> Option<&mut Digest> {
+        match self {
+            Self::H1(s) => Some(s.digest_mut()),
+            Self::H2(s) => s.digest_mut(),
+        }
+    }
+
     /// Return the server (peer) address of the connection.
     pub fn server_addr(&self) -> Option<&SocketAddr> {
         match self {
diff --git a/pingora-core/src/protocols/http/server.rs b/pingora-core/src/protocols/http/server.rs
index d4113b08b..c6479e7a0 100644
--- a/pingora-core/src/protocols/http/server.rs
+++ b/pingora-core/src/protocols/http/server.rs
@@ -25,6 +25,7 @@ use http::{header::AsHeaderName, HeaderMap};
 use log::error;
 use pingora_error::Result;
 use pingora_http::{RequestHeader, ResponseHeader};
+use std::time::Duration;
 
 /// HTTP server session object for both HTTP/1.x and HTTP/2
 pub enum Session {
@@ -188,6 +189,35 @@ impl Session {
         }
     }
 
+    /// Sets the downstream write timeout. This will trigger if we're unable
+    /// to write to the stream after `duration`. If a `min_send_rate` is
+    /// configured then the `min_send_rate` calculated timeout has higher priority.
+    ///
+    /// This is a noop for h2.
+    pub fn set_write_timeout(&mut self, timeout: Duration) {
+        match self {
+            Self::H1(s) => s.set_write_timeout(timeout),
+            Self::H2(_) => {}
+        }
+    }
+
+    /// Sets the minimum downstream send rate in bytes per second. This
+    /// is used to calculate a write timeout in seconds based on the size
+    /// of the buffer being written. If a `min_send_rate` is configured it
+    /// has higher priority over a set `write_timeout`. The minimum send
+    /// rate must be greater than zero.
+    ///
+    /// Calculated write timeout is guaranteed to be at least 1s if `min_send_rate`
+    /// is greater than zero, a send rate of zero is a noop.
+    ///
+    /// This is a noop for h2.
+    pub fn set_min_send_rate(&mut self, rate: usize) {
+        match self {
+            Self::H1(s) => s.set_min_send_rate(rate),
+            Self::H2(_) => {}
+        }
+    }
+
     /// Return a digest of the request including the method, path and Host header
     // TODO: make this use a `Formatter`
     pub fn request_summary(&self) -> String {
@@ -357,7 +387,7 @@ impl Session {
         }
     }
 
-    /// Return the digest for the session.
+    /// Return the [Digest] for the connection.
     pub fn digest(&self) -> Option<&Digest> {
         match self {
             Self::H1(s) => Some(s.digest()),
@@ -365,6 +395,16 @@ impl Session {
         }
     }
 
+    /// Return a mutable [Digest] reference for the connection.
+    ///
+    /// Will return `None` if multiple H2 streams are open.
+    pub fn digest_mut(&mut self) -> Option<&mut Digest> {
+        match self {
+            Self::H1(s) => Some(s.digest_mut()),
+            Self::H2(s) => s.digest_mut(),
+        }
+    }
+
     /// Return the client (peer) address of the connection.
     pub fn client_addr(&self) -> Option<&SocketAddr> {
         match self {
diff --git a/pingora-core/src/protocols/http/v1/client.rs b/pingora-core/src/protocols/http/v1/client.rs
index 7a72ed40b..8c2ab14ed 100644
--- a/pingora-core/src/protocols/http/v1/client.rs
+++ b/pingora-core/src/protocols/http/v1/client.rs
@@ -631,10 +631,19 @@ impl HttpSession {
         // TODO: support h1 trailer
     }
 
+    /// Return the [Digest] of the connection
+    ///
+    /// For reused connection, the timing in the digest will reflect its initial handshakes
+    /// The caller should check if the connection is reused to avoid misuse the timing field.
     pub fn digest(&self) -> &Digest {
         &self.digest
     }
 
+    /// Return a mutable [Digest] reference for the connection, see [`digest`] for more details.
+    pub fn digest_mut(&mut self) -> &mut Digest {
+        &mut self.digest
+    }
+
     /// Return the server (peer) address recorded in the connection digest.
     pub fn server_addr(&self) -> Option<&SocketAddr> {
         self.digest()
diff --git a/pingora-core/src/protocols/http/v1/server.rs b/pingora-core/src/protocols/http/v1/server.rs
index 909381c41..82a93a20f 100644
--- a/pingora-core/src/protocols/http/v1/server.rs
+++ b/pingora-core/src/protocols/http/v1/server.rs
@@ -72,6 +72,8 @@ pub struct HttpSession {
     upgraded: bool,
     /// Digest to track underlying connection metrics
     digest: Box<Digest>,
+    /// Minimum send rate to the client
+    min_send_rate: Option<usize>,
 }
 
 impl HttpSession {
@@ -106,6 +108,7 @@ impl HttpSession {
             retry_buffer: None,
             upgraded: false,
             digest,
+            min_send_rate: None,
         }
     }
 
@@ -511,6 +514,18 @@ impl HttpSession {
         is_buf_keepalive(self.get_header(header::CONNECTION))
     }
 
+    // calculate write timeout from min_send_rate if set, otherwise return write_timeout
+    fn write_timeout(&self, buf_len: usize) -> Option<Duration> {
+        let Some(min_send_rate) = self.min_send_rate.filter(|r| *r > 0) else {
+            return self.write_timeout;
+        };
+
+        // min timeout is 1s
+        let ms = (buf_len.max(min_send_rate) as f64 / min_send_rate as f64) * 1000.0;
+        // truncates unrealistically large values (we'll be out of memory before this happens)
+        Some(Duration::from_millis(ms as u64))
+    }
+
     /// Apply keepalive settings according to the client
     /// For HTTP 1.1, assume keepalive as long as there is no `Connection: Close` request header.
     /// For HTTP 1.0, only keepalive if there is an explicit header `Connection: keep-alive`.
@@ -579,7 +594,7 @@ impl HttpSession {
     /// to be written, e.g., writing more bytes than what the `Content-Length` header suggests
     pub async fn write_body(&mut self, buf: &[u8]) -> Result<Option<usize>> {
         // TODO: check if the response header is written
-        match self.write_timeout {
+        match self.write_timeout(buf.len()) {
             Some(t) => match timeout(t, self.do_write_body(buf)).await {
                 Ok(res) => res,
                 Err(_) => Error::e_explain(WriteTimedout, format!("writing body, timeout: {t:?}")),
@@ -588,7 +603,7 @@ impl HttpSession {
         }
     }
 
-    async fn write_body_buf(&mut self) -> Result<Option<usize>> {
+    async fn do_write_body_buf(&mut self) -> Result<Option<usize>> {
         // Don't flush empty chunks, they are considered end of body for chunks
         if self.body_write_buf.is_empty() {
             return Ok(None);
@@ -609,6 +624,16 @@ impl HttpSession {
         written
     }
 
+    async fn write_body_buf(&mut self) -> Result<Option<usize>> {
+        match self.write_timeout(self.body_write_buf.len()) {
+            Some(t) => match timeout(t, self.do_write_body_buf()).await {
+                Ok(res) => res,
+                Err(_) => Error::e_explain(WriteTimedout, format!("writing body, timeout: {t:?}")),
+            },
+            None => self.do_write_body_buf().await,
+        }
+    }
+
     fn maybe_force_close_body_reader(&mut self) {
         if self.upgraded && !self.body_reader.body_done() {
             // response is done, reset the request body to close
@@ -778,11 +803,37 @@ impl HttpSession {
         }
     }
 
+    /// Sets the downstream write timeout. This will trigger if we're unable
+    /// to write to the stream after `duration`. If a `min_send_rate` is
+    /// configured then the `min_send_rate` calculated timeout has higher priority.
+    pub fn set_write_timeout(&mut self, timeout: Duration) {
+        self.write_timeout = Some(timeout);
+    }
+
+    /// Sets the minimum downstream send rate in bytes per second. This
+    /// is used to calculate a write timeout in seconds based on the size
+    /// of the buffer being written. If a `min_send_rate` is configured it
+    /// has higher priority over a set `write_timeout`. The minimum send
+    /// rate must be greater than zero.
+    ///
+    /// Calculated write timeout is guaranteed to be at least 1s if `min_send_rate`
+    /// is greater than zero, a send rate of zero is a noop.
+    pub fn set_min_send_rate(&mut self, min_send_rate: usize) {
+        if min_send_rate > 0 {
+            self.min_send_rate = Some(min_send_rate);
+        }
+    }
+
     /// Return the [Digest] of the connection.
     pub fn digest(&self) -> &Digest {
         &self.digest
     }
 
+    /// Return a mutable [Digest] reference for the connection.
+    pub fn digest_mut(&mut self) -> &mut Digest {
+        &mut self.digest
+    }
+
     /// Return the client (peer) address of the underlying connection.
     pub fn client_addr(&self) -> Option<&SocketAddr> {
         self.digest()
@@ -1583,6 +1634,30 @@ mod tests_stream {
         assert!(written.is_none());
     }
 
+    #[tokio::test]
+    #[should_panic(expected = "There is still data left to write.")]
+    async fn test_write_body_buf_write_timeout() {
+        let wire1 = b"HTTP/1.1 200 OK\r\nContent-Length: 3\r\n\r\n";
+        let wire2 = b"abc";
+        let mock_io = Builder::new()
+            .write(wire1)
+            .wait(Duration::from_millis(500))
+            .write(wire2)
+            .build();
+        let mut http_stream = HttpSession::new(Box::new(mock_io));
+        http_stream.write_timeout = Some(Duration::from_millis(100));
+        let mut new_response = ResponseHeader::build(StatusCode::OK, None).unwrap();
+        new_response.append_header("Content-Length", "3").unwrap();
+        http_stream.update_resp_headers = false;
+        http_stream
+            .write_response_header_ref(&new_response)
+            .await
+            .unwrap();
+        http_stream.body_write_buf = BytesMut::from(&b"abc"[..]);
+        let res = http_stream.write_body_buf().await;
+        assert_eq!(res.unwrap_err().etype(), &WriteTimedout);
+    }
+
     #[tokio::test]
     async fn test_write_continue_resp() {
         let wire = b"HTTP/1.1 100 Continue\r\n\r\n";
@@ -1610,6 +1685,48 @@ mod tests_stream {
         response.set_version(http::Version::HTTP_11);
         assert!(!is_upgrade_resp(&response));
     }
+
+    #[test]
+    fn test_get_write_timeout() {
+        let mut http_stream = HttpSession::new(Box::new(Builder::new().build()));
+        let expected = Duration::from_secs(5);
+
+        http_stream.set_write_timeout(expected);
+        assert_eq!(Some(expected), http_stream.write_timeout(50));
+    }
+
+    #[test]
+    fn test_get_write_timeout_none() {
+        let http_stream = HttpSession::new(Box::new(Builder::new().build()));
+        assert!(http_stream.write_timeout(50).is_none());
+    }
+
+    #[test]
+    fn test_get_write_timeout_min_send_rate_zero_noop() {
+        let mut http_stream = HttpSession::new(Box::new(Builder::new().build()));
+        http_stream.set_min_send_rate(0);
+        assert!(http_stream.write_timeout(50).is_none());
+    }
+
+    #[test]
+    fn test_get_write_timeout_min_send_rate_overrides_write_timeout() {
+        let mut http_stream = HttpSession::new(Box::new(Builder::new().build()));
+        let expected = Duration::from_millis(29800);
+
+        http_stream.set_write_timeout(Duration::from_secs(60));
+        http_stream.set_min_send_rate(5000);
+
+        assert_eq!(Some(expected), http_stream.write_timeout(149000));
+    }
+
+    #[test]
+    fn test_get_write_timeout_min_send_rate_max_zero_buf() {
+        let mut http_stream = HttpSession::new(Box::new(Builder::new().build()));
+        let expected = Duration::from_secs(1);
+
+        http_stream.set_min_send_rate(1);
+        assert_eq!(Some(expected), http_stream.write_timeout(0));
+    }
 }
 
 #[cfg(test)]
diff --git a/pingora-core/src/protocols/http/v2/client.rs b/pingora-core/src/protocols/http/v2/client.rs
index 15f86d08e..86a3fe3da 100644
--- a/pingora-core/src/protocols/http/v2/client.rs
+++ b/pingora-core/src/protocols/http/v2/client.rs
@@ -305,11 +305,18 @@ impl Http2Session {
     /// Return the [Digest] of the connection
     ///
     /// For reused connection, the timing in the digest will reflect its initial handshakes
-    /// The caller should check if the connection is reused to avoid misuse the timing field
+    /// The caller should check if the connection is reused to avoid misuse the timing field.
     pub fn digest(&self) -> Option<&Digest> {
         Some(self.conn.digest())
     }
 
+    /// Return a mutable [Digest] reference for the connection, see [`digest`] for more details.
+    ///
+    /// Will return `None` if multiple H2 streams are open.
+    pub fn digest_mut(&mut self) -> Option<&mut Digest> {
+        self.conn.digest_mut()
+    }
+
     /// Return the server (peer) address recorded in the connection digest.
     pub fn server_addr(&self) -> Option<&SocketAddr> {
         self.conn
diff --git a/pingora-core/src/protocols/http/v2/server.rs b/pingora-core/src/protocols/http/v2/server.rs
index 57238d107..f9072d6b6 100644
--- a/pingora-core/src/protocols/http/v2/server.rs
+++ b/pingora-core/src/protocols/http/v2/server.rs
@@ -454,6 +454,11 @@ impl HttpSession {
         Some(&self.digest)
     }
 
+    /// Return a mutable [Digest] reference for the connection.
+    pub fn digest_mut(&mut self) -> Option<&mut Digest> {
+        Arc::get_mut(&mut self.digest)
+    }
+
     /// Return the server (local) address recorded in the connection digest.
     pub fn server_addr(&self) -> Option<&SocketAddr> {
         self.digest.socket_digest.as_ref().map(|d| d.local_addr())?
diff --git a/pingora-core/src/protocols/l4/socket.rs b/pingora-core/src/protocols/l4/socket.rs
index 186fbecb3..93d334ed5 100644
--- a/pingora-core/src/protocols/l4/socket.rs
+++ b/pingora-core/src/protocols/l4/socket.rs
@@ -15,6 +15,7 @@
 //! Generic socket type
 
 use crate::{Error, OrErr};
+use log::warn;
 use nix::sys::socket::{getpeername, getsockname, SockaddrStorage};
 use std::cmp::Ordering;
 use std::hash::{Hash, Hasher};
@@ -174,14 +175,23 @@ impl std::str::FromStr for SocketAddr {
     type Err = Box<Error>;
 
     // This is very basic parsing logic, it might treat invalid IP:PORT str as UDS path
-    // TODO: require UDS to have some prefix
     fn from_str(s: &str) -> Result<Self, Self::Err> {
-        match StdSockAddr::from_str(s) {
-            Ok(addr) => Ok(SocketAddr::Inet(addr)),
-            Err(_) => {
-                let uds_socket = StdUnixSockAddr::from_pathname(s)
-                    .or_err(crate::BindError, "invalid UDS path")?;
-                Ok(SocketAddr::Unix(uds_socket))
+        if s.starts_with("unix:") {
+            // format unix:/tmp/server.socket
+            let path = s.trim_start_matches("unix:");
+            let uds_socket = StdUnixSockAddr::from_pathname(path)
+                .or_err(crate::BindError, "invalid UDS path")?;
+            Ok(SocketAddr::Unix(uds_socket))
+        } else {
+            match StdSockAddr::from_str(s) {
+                Ok(addr) => Ok(SocketAddr::Inet(addr)),
+                Err(_) => {
+                    // Try to parse as UDS for backward compatibility
+                    let uds_socket = StdUnixSockAddr::from_pathname(s)
+                        .or_err(crate::BindError, "invalid UDS path")?;
+                    warn!("Raw Unix domain socket path support will be deprecated, add 'unix:' prefix instead");
+                    Ok(SocketAddr::Unix(uds_socket))
+                }
             }
         }
     }
@@ -246,4 +256,10 @@ mod test {
         let uds: SocketAddr = "/tmp/my.sock".parse().unwrap();
         assert!(uds.as_unix().is_some());
     }
+
+    #[test]
+    fn parse_uds_with_prefix() {
+        let uds: SocketAddr = "unix:/tmp/my.sock".parse().unwrap();
+        assert!(uds.as_unix().is_some());
+    }
 }
diff --git a/pingora-core/src/server/configuration/mod.rs b/pingora-core/src/server/configuration/mod.rs
index c118c84ac..5174d4072 100644
--- a/pingora-core/src/server/configuration/mod.rs
+++ b/pingora-core/src/server/configuration/mod.rs
@@ -119,7 +119,7 @@ impl Default for ServerConf {
 /// Command-line options
 ///
 /// Call `Opt::from_args()` to build this object from the process's command line arguments.
-#[derive(Parser, Debug)]
+#[derive(Parser, Debug, Default)]
 #[clap(name = "basic", long_about = None)]
 pub struct Opt {
     /// Whether this server should try to upgrade from a running old server
@@ -163,15 +163,6 @@ pub struct Opt {
     pub conf: Option<String>,
 }
 
-/// Create the default instance of Opt based on the current command-line args.
-/// This is equivalent to running `Opt::parse` but does not require the
-/// caller to have included the `clap::Parser`
-impl Default for Opt {
-    fn default() -> Self {
-        Opt::parse()
-    }
-}
-
 impl ServerConf {
     // Does not has to be async until we want runtime reload
     pub fn load_from_yaml<P>(path: P) -> Result<Self>
@@ -236,6 +227,15 @@ impl ServerConf {
     }
 }
 
+/// Create an instance of Opt by parsing the current command-line args.
+/// This is equivalent to running `Opt::parse` but does not require the
+/// caller to have included the `clap::Parser`
+impl Opt {
+    pub fn parse_args() -> Self {
+        Opt::parse()
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
diff --git a/pingora-header-serde/src/lib.rs b/pingora-header-serde/src/lib.rs
index 73b9b2929..e78897a78 100644
--- a/pingora-header-serde/src/lib.rs
+++ b/pingora-header-serde/src/lib.rs
@@ -131,7 +131,7 @@ fn resp_header_to_buf(resp: &ResponseHeader, buf: &mut Vec<u8>) -> usize {
 }
 
 // Should match pingora http1 setting
-const MAX_HEADERS: usize = 160;
+const MAX_HEADERS: usize = 256;
 
 #[inline]
 fn buf_to_http_header(buf: &[u8]) -> Result<ResponseHeader> {
diff --git a/pingora-memory-cache/src/lib.rs b/pingora-memory-cache/src/lib.rs
index 5e0254bd9..b6e78bda0 100644
--- a/pingora-memory-cache/src/lib.rs
+++ b/pingora-memory-cache/src/lib.rs
@@ -45,6 +45,14 @@ impl CacheStatus {
             Self::LockHit => "lock_hit",
         }
     }
+
+    /// Returns whether this status represents a cache hit.
+    pub fn is_hit(&self) -> bool {
+        match self {
+            CacheStatus::Hit | CacheStatus::LockHit => true,
+            CacheStatus::Miss | CacheStatus::Expired => false,
+        }
+    }
 }
 
 #[derive(Debug, Clone)]
diff --git a/pingora-openssl/src/lib.rs b/pingora-openssl/src/lib.rs
index 25f7cb715..7d52d683d 100644
--- a/pingora-openssl/src/lib.rs
+++ b/pingora-openssl/src/lib.rs
@@ -25,6 +25,7 @@ pub use tokio_openssl as tokio_ssl;
 pub mod ext;
 
 // export commonly used libs
+pub use ssl_lib::dh;
 pub use ssl_lib::error;
 pub use ssl_lib::hash;
 pub use ssl_lib::nid;
diff --git a/pingora-proxy/src/lib.rs b/pingora-proxy/src/lib.rs
index 0add56aba..99e7d6d06 100644
--- a/pingora-proxy/src/lib.rs
+++ b/pingora-proxy/src/lib.rs
@@ -110,6 +110,14 @@ impl<SV> HttpProxy<SV> {
         }
     }
 
+    fn handle_init_modules(&mut self)
+    where
+        SV: ProxyHttp,
+    {
+        self.inner
+            .init_downstream_modules(&mut self.downstream_modules);
+    }
+
     async fn handle_new_request(
         &self,
         mut downstream_session: Box<HttpSession>,
@@ -739,7 +747,10 @@ use pingora_core::services::listening::Service;
 /// Create a [Service] from the user implemented [ProxyHttp].
 ///
 /// The returned [Service] can be hosted by a [pingora_core::server::Server] directly.
-pub fn http_proxy_service<SV>(conf: &Arc<ServerConf>, inner: SV) -> Service<HttpProxy<SV>> {
+pub fn http_proxy_service<SV>(conf: &Arc<ServerConf>, inner: SV) -> Service<HttpProxy<SV>>
+where
+    SV: ProxyHttp,
+{
     http_proxy_service_with_name(conf, inner, "Pingora HTTP Proxy Service")
 }
 
@@ -750,11 +761,11 @@ pub fn http_proxy_service_with_name<SV>(
     conf: &Arc<ServerConf>,
     inner: SV,
     name: &str,
-) -> Service<HttpProxy<SV>> {
+) -> Service<HttpProxy<SV>>
+where
+    SV: ProxyHttp,
+{
     let mut proxy = HttpProxy::new(inner, conf.clone());
-    // Add disabled downstream compression module by default
-    proxy
-        .downstream_modules
-        .add_module(ResponseCompressionBuilder::enable(0));
+    proxy.handle_init_modules();
     Service::new(name.to_string(), proxy)
 }
diff --git a/pingora-proxy/src/proxy_h1.rs b/pingora-proxy/src/proxy_h1.rs
index feac860c5..de49c03c4 100644
--- a/pingora-proxy/src/proxy_h1.rs
+++ b/pingora-proxy/src/proxy_h1.rs
@@ -98,10 +98,7 @@ impl<SV> HttpProxy<SV> {
         );
 
         match ret {
-            Ok((_first, _second)) => {
-                client_session.respect_keepalive();
-                (true, true, None)
-            }
+            Ok((_first, _second)) => (true, true, None),
             Err(e) => (false, false, Some(e)),
         }
     }
diff --git a/pingora-proxy/src/proxy_trait.rs b/pingora-proxy/src/proxy_trait.rs
index 6f429ac3e..2d99567eb 100644
--- a/pingora-proxy/src/proxy_trait.rs
+++ b/pingora-proxy/src/proxy_trait.rs
@@ -40,6 +40,17 @@ pub trait ProxyHttp {
         ctx: &mut Self::CTX,
     ) -> Result<Box<HttpPeer>>;
 
+    /// Set up downstream modules.
+    ///
+    /// In this phase, users can add or configure [HttpModules] before the server starts up.
+    ///
+    /// In the default implementation of this method, [ResponseCompressionBuilder] is added
+    /// and disabled.
+    fn init_downstream_modules(&self, modules: &mut HttpModules) {
+        // Add disabled downstream compression module by default
+        modules.add_module(ResponseCompressionBuilder::enable(0));
+    }
+
     /// Handle the incoming request.
     ///
     /// In this phase, users can parse, validate, rate limit, perform access control and/or
diff --git a/tinyufo/src/estimation.rs b/tinyufo/src/estimation.rs
index 6f165ac0e..11b9d923d 100644
--- a/tinyufo/src/estimation.rs
+++ b/tinyufo/src/estimation.rs
@@ -36,23 +36,36 @@ impl Estimator {
 
     fn optimal(items: usize) -> Self {
         let (slots, hashes) = Self::optimal_paras(items);
-        Self::new(hashes, slots)
+        Self::new(hashes, slots, RandomState::new)
     }
 
     fn compact(items: usize) -> Self {
         let (slots, hashes) = Self::optimal_paras(items / 100);
-        Self::new(hashes, slots)
+        Self::new(hashes, slots, RandomState::new)
     }
 
-    /// Create a new `Estimator` with the given amount of hashes and columns (slots).
-    pub fn new(hashes: usize, slots: usize) -> Self {
+    #[cfg(test)]
+    fn seeded(items: usize) -> Self {
+        let (slots, hashes) = Self::optimal_paras(items);
+        Self::new(hashes, slots, || RandomState::with_seeds(2, 3, 4, 5))
+    }
+
+    #[cfg(test)]
+    fn seeded_compact(items: usize) -> Self {
+        let (slots, hashes) = Self::optimal_paras(items / 100);
+        Self::new(hashes, slots, || RandomState::with_seeds(2, 3, 4, 5))
+    }
+
+    /// Create a new `Estimator` with the given amount of hashes and columns (slots) using
+    /// the given random source.
+    pub fn new(hashes: usize, slots: usize, random: impl Fn() -> RandomState) -> Self {
         let mut estimator = Vec::with_capacity(hashes);
         for _ in 0..hashes {
             let mut slot = Vec::with_capacity(slots);
             for _ in 0..slots {
                 slot.push(AtomicU8::new(0));
             }
-            estimator.push((slot.into_boxed_slice(), RandomState::new()));
+            estimator.push((slot.into_boxed_slice(), random()));
         }
 
         Estimator {
@@ -161,6 +174,26 @@ impl TinyLfu {
             window_limit: cache_size * 8,
         }
     }
+
+    #[cfg(test)]
+    pub fn new_seeded(cache_size: usize) -> Self {
+        Self {
+            estimator: Estimator::seeded(cache_size),
+            window_counter: Default::default(),
+            // 8x: just a heuristic to balance the memory usage and accuracy
+            window_limit: cache_size * 8,
+        }
+    }
+
+    #[cfg(test)]
+    pub fn new_compact_seeded(cache_size: usize) -> Self {
+        Self {
+            estimator: Estimator::seeded_compact(cache_size),
+            window_counter: Default::default(),
+            // 8x: just a heuristic to balance the memory usage and accuracy
+            window_limit: cache_size * 8,
+        }
+    }
 }
 
 #[cfg(test)]
diff --git a/tinyufo/src/lib.rs b/tinyufo/src/lib.rs
index 805373e81..eb4cbaf48 100644
--- a/tinyufo/src/lib.rs
+++ b/tinyufo/src/lib.rs
@@ -473,7 +473,9 @@ mod tests {
 
     #[test]
     fn test_evict_from_small() {
-        let cache = TinyUfo::new(5, 5);
+        let mut cache = TinyUfo::new(5, 5);
+        cache.random_status = RandomState::with_seeds(2, 3, 4, 5);
+        cache.queues.estimator = TinyLfu::new_seeded(5);
 
         cache.put(1, 1, 1);
         cache.put(2, 2, 2);
@@ -496,7 +498,9 @@ mod tests {
 
     #[test]
     fn test_evict_from_small_to_main() {
-        let cache = TinyUfo::new(5, 5);
+        let mut cache = TinyUfo::new(5, 5);
+        cache.random_status = RandomState::with_seeds(2, 3, 4, 5);
+        cache.queues.estimator = TinyLfu::new_seeded(5);
 
         cache.put(1, 1, 1);
         cache.put(2, 2, 2);
@@ -510,20 +514,30 @@ mod tests {
         assert_eq!(cache.peek_queue(2), Some(SMALL));
         assert_eq!(cache.peek_queue(3), Some(SMALL));
 
-        let evicted = cache.put(4, 4, 1);
+        let evicted = cache.put(4, 4, 2);
         assert_eq!(evicted.len(), 1);
-        assert_eq!(evicted[0].data, 2);
+        assert_eq!(evicted[0].weight, 2);
 
         assert_eq!(cache.peek_queue(1), Some(MAIN));
-        // 2 is evicted because 1 is in main
-        assert_eq!(cache.peek_queue(2), None);
-        assert_eq!(cache.peek_queue(3), Some(SMALL));
-        assert_eq!(cache.peek_queue(4), Some(SMALL));
+        // either 2, 3, or 4 was evicted. Check evicted for which.
+        let mut remaining = vec![2, 3, 4];
+        remaining.remove(
+            remaining
+                .iter()
+                .position(|x| *x == evicted[0].data)
+                .unwrap(),
+        );
+        assert_eq!(cache.peek_queue(evicted[0].key), None);
+        for k in remaining {
+            assert_eq!(cache.peek_queue(k), Some(SMALL));
+        }
     }
 
     #[test]
     fn test_evict_reentry() {
-        let cache = TinyUfo::new(5, 5);
+        let mut cache = TinyUfo::new(5, 5);
+        cache.random_status = RandomState::with_seeds(2, 3, 4, 5);
+        cache.queues.estimator = TinyLfu::new_seeded(5);
 
         cache.put(1, 1, 1);
         cache.put(2, 2, 2);
@@ -555,7 +569,9 @@ mod tests {
 
     #[test]
     fn test_evict_entry_denied() {
-        let cache = TinyUfo::new(5, 5);
+        let mut cache = TinyUfo::new(5, 5);
+        cache.random_status = RandomState::with_seeds(2, 3, 4, 5);
+        cache.queues.estimator = TinyLfu::new_seeded(5);
 
         cache.put(1, 1, 1);
         cache.put(2, 2, 2);
@@ -583,7 +599,9 @@ mod tests {
 
     #[test]
     fn test_force_put() {
-        let cache = TinyUfo::new(5, 5);
+        let mut cache = TinyUfo::new(5, 5);
+        cache.random_status = RandomState::with_seeds(2, 3, 4, 5);
+        cache.queues.estimator = TinyLfu::new_seeded(5);
 
         cache.put(1, 1, 1);
         cache.put(2, 2, 2);
@@ -612,7 +630,9 @@ mod tests {
 
     #[test]
     fn test_evict_from_main() {
-        let cache = TinyUfo::new(5, 5);
+        let mut cache = TinyUfo::new(5, 5);
+        cache.random_status = RandomState::with_seeds(2, 3, 4, 5);
+        cache.queues.estimator = TinyLfu::new_seeded(5);
 
         cache.put(1, 1, 1);
         cache.put(2, 2, 2);
@@ -649,7 +669,9 @@ mod tests {
 
     #[test]
     fn test_evict_from_small_compact() {
-        let cache = TinyUfo::new(5, 5);
+        let mut cache = TinyUfo::new(5, 5);
+        cache.random_status = RandomState::with_seeds(2, 3, 4, 5);
+        cache.queues.estimator = TinyLfu::new_compact_seeded(5);
 
         cache.put(1, 1, 1);
         cache.put(2, 2, 2);
@@ -672,7 +694,9 @@ mod tests {
 
     #[test]
     fn test_evict_from_small_to_main_compact() {
-        let cache = TinyUfo::new(5, 5);
+        let mut cache = TinyUfo::new(5, 5);
+        cache.random_status = RandomState::with_seeds(2, 3, 4, 5);
+        cache.queues.estimator = TinyLfu::new_compact_seeded(5);
 
         cache.put(1, 1, 1);
         cache.put(2, 2, 2);


Test patch:
diff --git a/pingora-proxy/tests/test_upstream.rs b/pingora-proxy/tests/test_upstream.rs
index ad7aaaf38..260083289 100644
--- a/pingora-proxy/tests/test_upstream.rs
+++ b/pingora-proxy/tests/test_upstream.rs
@@ -133,6 +133,60 @@ async fn test_ws_server_ends_conn() {
     assert!(ws_stream.next().await.is_none());
 }
 
+#[tokio::test]
+async fn test_download_timeout() {
+    init();
+    use hyper::body::HttpBody;
+    use tokio::time::sleep;
+
+    let client = hyper::Client::new();
+    let uri: hyper::Uri = "http://127.0.0.1:6147/download/".parse().unwrap();
+    let req = hyper::Request::builder()
+        .uri(uri)
+        .header("x-write-timeout", "1")
+        .body(hyper::Body::empty())
+        .unwrap();
+    let mut res = client.request(req).await.unwrap();
+    assert_eq!(res.status(), StatusCode::OK);
+
+    let mut err = false;
+    sleep(Duration::from_secs(2)).await;
+    while let Some(chunk) = res.body_mut().data().await {
+        if chunk.is_err() {
+            err = true;
+        }
+    }
+    assert!(err);
+}
+
+#[tokio::test]
+async fn test_download_timeout_min_rate() {
+    init();
+    use hyper::body::HttpBody;
+    use tokio::time::sleep;
+
+    let client = hyper::Client::new();
+    let uri: hyper::Uri = "http://127.0.0.1:6147/download/".parse().unwrap();
+    let req = hyper::Request::builder()
+        .uri(uri)
+        .header("x-write-timeout", "1")
+        .header("x-min-rate", "10000")
+        .body(hyper::Body::empty())
+        .unwrap();
+    let mut res = client.request(req).await.unwrap();
+    assert_eq!(res.status(), StatusCode::OK);
+
+    let mut err = false;
+    sleep(Duration::from_secs(2)).await;
+    while let Some(chunk) = res.body_mut().data().await {
+        if chunk.is_err() {
+            err = true;
+        }
+    }
+    // no error as write timeout is overridden by min rate
+    assert!(!err);
+}
+
 mod test_cache {
     use super::*;
     use std::str::FromStr;
diff --git a/pingora-proxy/tests/utils/conf/origin/conf/nginx.conf b/pingora-proxy/tests/utils/conf/origin/conf/nginx.conf
index 07e57e8c2..a41a743dd 100644
--- a/pingora-proxy/tests/utils/conf/origin/conf/nginx.conf
+++ b/pingora-proxy/tests/utils/conf/origin/conf/nginx.conf
@@ -296,6 +296,15 @@ http {
             }
         }
 
+        location /download/ {
+            content_by_lua_block {
+                ngx.req.read_body()
+                local body = string.rep("A", 4194304)
+                ngx.header["Content-Length"] = #body
+                ngx.print(body)
+            }
+        }
+
         location /tls_verify {
             keepalive_timeout 0;
             return 200;
diff --git a/pingora-proxy/tests/utils/server_utils.rs b/pingora-proxy/tests/utils/server_utils.rs
index 7b20630dd..ec1a96277 100644
--- a/pingora-proxy/tests/utils/server_utils.rs
+++ b/pingora-proxy/tests/utils/server_utils.rs
@@ -39,6 +39,7 @@ use pingora_proxy::{ProxyHttp, Session};
 use std::collections::{HashMap, HashSet};
 use std::sync::Arc;
 use std::thread;
+use std::time::Duration;
 
 pub struct ExampleProxyHttps {}
 
@@ -230,6 +231,17 @@ impl ProxyHttp for ExampleProxyHttp {
 
     async fn request_filter(&self, session: &mut Session, _ctx: &mut Self::CTX) -> Result<bool> {
         let req = session.req_header();
+
+        let write_timeout = req
+            .headers
+            .get("x-write-timeout")
+            .and_then(|v| v.to_str().ok().and_then(|v| v.parse().ok()));
+
+        let min_rate = req
+            .headers
+            .get("x-min-rate")
+            .and_then(|v| v.to_str().ok().and_then(|v| v.parse().ok()));
+
         let downstream_compression = req.headers.get("x-downstream-compression").is_some();
         if !downstream_compression {
             // enable upstream compression for all requests by default
@@ -242,6 +254,13 @@ impl ProxyHttp for ExampleProxyHttp {
                 .adjust_level(0);
         }
 
+        if let Some(min_rate) = min_rate {
+            session.set_min_send_rate(min_rate);
+        }
+        if let Some(write_timeout) = write_timeout {
+            session.set_write_timeout(Duration::from_secs(write_timeout));
+        }
+
         Ok(false)
     }