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.

.bleep

@@ -1 +1 @@
-dc97a9520b124eb464f348b0381991d8669c8d8a
+8e05a8f5b9d09885e6374011c422678043a2bda0

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(

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(