Repository: deno
Issue: `fetch()` fails when reusing persistent connection that has closed.
Labels: bug,ext/fetch

I've done some more investigation since #11495 and believe I've found the issue.

If a server unexpectedly closes a connection during a call to `fetch()`, the *next* call to `fetch()` will fail.

This seems like a bug for 2 reasons:

1. Deno should probably check that the existing persistent connection is still valid (connected) before attempting to re-use it.
2. This behavior makes it really difficult to debug which fetch() is causing an error.

Steps to reproduce
----------------------

Run this code against a server that closes a connection early.  

```typescript


const port = 9999

const brokenURL = `http://127.0.0.1:${port}/`

// If the server correctly reads all of the body, then there's no problem:
const fixedURL = `http://127.0.0.1:${port}/?drain=true`

// Bigger files seem to trigger this more reliably:
const bigFile = new Uint8Array(1024*1024*50)

async function get(): Promise<void> {
    // GETs always work OK because there's no body.
    const r = await fetch(brokenURL)
    if (!r.ok) throw {result: r, status: r.status, statusText: r.statusText}
}

async function put(url: string) {
    const r = await fetch(url, { method: "PUT", body: bigFile })
    if (!r.ok) throw {result: r, status: r.status, statusText: r.statusText}
    console.log("Put finished")
}

// These work fine:
await get()
await get()
await get()
await get()
await get()
await get()

// This causes the keepalive connection to error, but in such a way that this
// call to Deno's fetch() still succeeds
await put(brokenURL)

// This tries to reuse the keepalive connection that has been closed and dies.
// IMO, this is a bug. As with a connection pool, it's best to make sure the
// existing connection is still valid before trying to reuse it.
await get()
```

I've reproduced this behavior in Actix-Web and Tide.   Yes, it's likely best behavior to always read the body, but this is an easy error to make and likely a case that folks writing Deno scripts will run into again. I'm hoping to spare them my headaches. :)


### Actix-Web ###

```rust

use std::io;

use actix_web::{App, HttpResponse, HttpServer, web::{Payload, Query, route}};
use futures::StreamExt;
use serde::Deserialize;

/// Responds to an HTTP request before the full body has been read. 
/// In Actix-web, this causes the conneciton to be dropped before the
/// client expects, which causes difficult-to-debug behavior in Deno's
/// fetch().
///
async fn fast_responder(
    Query(params): Query<Params>,
    mut req_body: Payload,
) -> HttpResponse {

    // If I wait for the full body *before* sending a response, things work:
    if params.drain.unwrap_or(false) {
        while req_body.next().await.is_some() {}
    }

    HttpResponse::Ok().finish()
}

fn main() -> io::Result<()>{
    let port = 9999;
    let server = HttpServer::new(|| {
        App::new().configure(|cfg| {
            cfg.route("/", route().to(fast_responder));
        })
    }).bind(format!("127.0.0.1:{}", port))?;

    let mut system = actix_web::rt::System::new("web server");
    system.block_on(server.run())?;

    Ok(())
}

#[derive(Deserialize)]
struct Params {
    drain: Option<bool>
}
```

### Tide ###

```rust
use futures::AsyncReadExt;
use tide::Request;
use tide::prelude::*;

async fn fast_responder(mut req: Request<()>) -> tide::Result {
    let params: Params = req.query()?;

    if params.drain.unwrap_or(false) {
        let mut reader  = req.take_body().into_reader();
        let mut buf = [0u8; 1024*16];
        while reader.read(&mut buf).await? > 0 {}
    }

    Ok("".into())
}

#[async_std::main]
async fn main() -> tide::Result<()> {
    let mut app = tide::new();
    app.at("/").get(fast_responder).put(fast_responder);
    app.listen("127.0.0.1:9999").await?;
    Ok(())
}

#[derive(Deserialize)]
struct Params {
    drain: Option<bool>
}
```