Don't reuse session when downstream errors during cache miss

When a downstream error occurs while a cache miss is being served, the
downstream error is ignored so the upstream can continue writing to the
cache. However, it is not safe to try reusing the server session in this
case.

3 files changed, 29 insertions, 21 deletions

diff --git a/.bleep b/.bleep
index 9840797..da57c1f 100644
--- a/.bleep
+++ b/.bleep
@@ -1 +1 @@
-dc97a9520b124eb464f348b0381991d8669c8d8a
\ No newline at end of file
+8e05a8f5b9d09885e6374011c422678043a2bda0
\ No newline at end of file
diff --git a/pingora-proxy/src/proxy_h1.rs b/pingora-proxy/src/proxy_h1.rs
index 15ce038..544a74e 100644
--- a/pingora-proxy/src/proxy_h1.rs
+++ b/pingora-proxy/src/proxy_h1.rs
@@ -98,7 +98,7 @@ impl<SV> HttpProxy<SV> {
         );
 
         match ret {
-            Ok((_first, _second)) => (true, true, None),
+            Ok((downstream_can_reuse, _upstream)) => (downstream_can_reuse, true, None),
             Err(e) => (false, false, Some(e)),
         }
     }
@@ -204,13 +204,14 @@ impl<SV> HttpProxy<SV> {
     }
 
     // todo use this function to replace bidirection_1to2()
+    // returns whether this server (downstream) session can be reused
     async fn proxy_handle_downstream(
         &self,
         session: &mut Session,
         tx: mpsc::Sender<HttpTask>,
         mut rx: mpsc::Receiver<HttpTask>,
         ctx: &mut SV::CTX,
-    ) -> Result<()>
+    ) -> Result<bool>
     where
         SV: ProxyHttp + Send + Sync,
         SV::CTX: Send + Sync,
@@ -416,16 +417,19 @@ impl<SV> HttpProxy<SV> {
             }
         }
 
-        match session.as_mut().finish_body().await {
-            Ok(_) => {
-                debug!("finished sending body to downstream");
-            }
-            Err(e) => {
-                error!("Error finish sending body to downstream: {}", e);
-                // TODO: don't do downstream keepalive
+        let mut reuse_downstream = !downstream_state.is_errored();
+        if reuse_downstream {
+            match session.as_mut().finish_body().await {
+                Ok(_) => {
+                    debug!("finished sending body to downstream");
+                }
+                Err(e) => {
+                    error!("Error finish sending body to downstream: {}", e);
+                    reuse_downstream = false;
+                }
             }
         }
-        Ok(())
+        Ok(reuse_downstream)
     }
 
     async fn h1_response_filter(
diff --git a/pingora-proxy/src/proxy_h2.rs b/pingora-proxy/src/proxy_h2.rs
index f133c2f..53b2c14 100644
--- a/pingora-proxy/src/proxy_h2.rs
+++ b/pingora-proxy/src/proxy_h2.rs
@@ -176,7 +176,7 @@ impl<SV> HttpProxy<SV> {
         );
 
         match ret {
-            Ok((_first, _second)) => (true, None),
+            Ok((downstream_can_reuse, _upstream)) => (downstream_can_reuse, None),
             Err(e) => (false, Some(e)),
         }
     }
@@ -212,13 +212,14 @@ impl<SV> HttpProxy<SV> {
         (server_session_reuse, error)
     }
 
+    // returns whether server (downstream) session can be reused
     async fn bidirection_1to2(
         &self,
         session: &mut Session,
         client_body: &mut h2::SendStream<bytes::Bytes>,
         mut rx: mpsc::Receiver<HttpTask>,
         ctx: &mut SV::CTX,
-    ) -> Result<()>
+    ) -> Result<bool>
     where
         SV: ProxyHttp + Send + Sync,
         SV::CTX: Send + Sync,
@@ -369,16 +370,19 @@ impl<SV> HttpProxy<SV> {
             }
         }
 
-        match session.as_mut().finish_body().await {
-            Ok(_) => {
-                debug!("finished sending body to downstream");
-            }
-            Err(e) => {
-                error!("Error finish sending body to downstream: {}", e);
-                // TODO: don't do downstream keepalive
+        let mut reuse_downstream = !downstream_state.is_errored();
+        if reuse_downstream {
+            match session.as_mut().finish_body().await {
+                Ok(_) => {
+                    debug!("finished sending body to downstream");
+                }
+                Err(e) => {
+                    error!("Error finish sending body to downstream: {}", e);
+                    reuse_downstream = false;
+                }
             }
         }
-        Ok(())
+        Ok(reuse_downstream)
     }
 
     async fn h2_response_filter(