fix(http1): return error if user body ends prematurely

- update proto::h1::end_body to return Result<()>
- update Encoder::end to return Error(NotEof) only when there's Content-length left to be addressed

Closes #2263

src/proto/h1/conn.rs

@@ -12,6 +12,7 @@ use super::{Decoder, Encode, EncodedBuf, Encoder, Http1Transaction, ParseContext
 use crate::common::{task, Pin, Poll, Unpin};
 use crate::headers::connection_keep_alive;
 use crate::proto::{BodyLength, DecodedLength, MessageHead};
+use crate::Result;
 
 const H2_PREFACE: &[u8] = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n";
 
@@ -584,7 +585,7 @@ where
         self.state.writing = state;
     }
 
-    pub fn end_body(&mut self) {
+    pub fn end_body(&mut self) -> Result<()> {
         debug_assert!(self.can_write_body());
 
         let state = match self.state.writing {
@@ -601,13 +602,18 @@ where
                             Writing::KeepAlive
                         }
                     }
-                    Err(_not_eof) => Writing::Closed,
+                    Err(_not_eof) => {
+                        return Err(crate::Error::new_user_body(
+                            crate::Error::new_body_write_aborted(),
+                        ))
+                    }
                 }
             }
-            _ => return,
+            _ => return Ok(()),
         };
 
         self.state.writing = state;
+        Ok(())
     }
 
     // When we get a parse error, depending on what side we are, we might be able

src/proto/h1/dispatch.rs

@@ -338,7 +338,7 @@ where
                             *clear_body = true;
                             if chunk.remaining() == 0 {
                                 trace!("discarding empty chunk");
-                                self.conn.end_body();
+                                self.conn.end_body()?;
                             } else {
                                 self.conn.write_body_and_end(chunk);
                             }
@@ -351,7 +351,7 @@ where
                         }
                     } else {
                         *clear_body = true;
-                        self.conn.end_body();
+                        self.conn.end_body()?;
                     }
                 } else {
                     return Poll::Pending;

src/proto/h1/encode.rs

@@ -87,7 +87,8 @@ impl Encoder {
             Kind::Chunked => Ok(Some(EncodedBuf {
                 kind: BufKind::ChunkedEnd(b"0\r\n\r\n"),
             })),
-            _ => Err(NotEof),
+            Kind::CloseDelimited => Ok(None),
+            Kind::Length(_) => Err(NotEof),
         }
     }
 
@@ -405,7 +406,7 @@ mod tests {
 
         assert_eq!(dst, b"foo bar");
         assert!(!encoder.is_eof());
-        encoder.end::<()>().unwrap_err();
+        encoder.end::<()>().unwrap();
 
         let msg2 = b"baz".as_ref();
         let buf2 = encoder.encode(msg2);
@@ -413,6 +414,6 @@ mod tests {
 
         assert_eq!(dst, b"foo barbaz");
         assert!(!encoder.is_eof());
-        encoder.end::<()>().unwrap_err();
+        encoder.end::<()>().unwrap();
     }
 }