std(http/server): close open connections on server close (#3679)

Due to structure of "Server" for each open connection there's a pending "read" op. Because connection owned by "Server" are not tracked, calling "Server.close()" doesn't close open connections.

This commit introduces simple tracking of connections for server and ensures owned connections are closed on "Server.close()".

std/http/server.ts

@@ -124,12 +124,23 @@ export class ServerRequest {
 
 export class Server implements AsyncIterable<ServerRequest> {
   private closing = false;
+  private connections: Conn[] = [];
 
   constructor(public listener: Listener) {}
 
   close(): void {
     this.closing = true;
     this.listener.close();
+    for (const conn of this.connections) {
+      try {
+        conn.close();
+      } catch (e) {
+        // Connection might have been already closed
+        if (!(e instanceof Deno.errors.BadResource)) {
+          throw e;
+        }
+      }
+    }
   }
 
   // Yields all HTTP requests on a single TCP connection.
@@ -162,6 +173,7 @@ export class Server implements AsyncIterable<ServerRequest> {
         // Something bad happened during response.
         // (likely other side closed during pipelined req)
         // req.done implies this connection already closed, so we can just return.
+        this.untrackConnection(req.conn);
         return;
       }
       // Consume unread body and trailers if receiver didn't consume those data
@@ -186,7 +198,23 @@ export class Server implements AsyncIterable<ServerRequest> {
       // TODO(ry): send a back a HTTP 503 Service Unavailable status.
     }
 
-    conn.close();
+    this.untrackConnection(conn);
+    try {
+      conn.close();
+    } catch (e) {
+      // might have been already closed
+    }
+  }
+
+  private trackConnection(conn: Conn): void {
+    this.connections.push(conn);
+  }
+
+  private untrackConnection(conn: Conn): void {
+    const index = this.connections.indexOf(conn);
+    if (index !== -1) {
+      this.connections.splice(index, 1);
+    }
   }
 
   // Accepts a new TCP connection and yields all HTTP requests that arrive on
@@ -207,6 +235,7 @@ export class Server implements AsyncIterable<ServerRequest> {
       }
       throw error;
     }
+    this.trackConnection(conn);
     // Try to accept another connection and add it to the multiplexer.
     mux.add(this.acceptConnAndIterateHttpRequests(mux));
     // Yield the requests that arrive on the just-accepted connection.

std/http/server_test.ts

@@ -6,8 +6,13 @@
 // https://github.com/golang/go/blob/master/src/net/http/responsewrite_test.go
 
 import { TextProtoReader } from "../textproto/mod.ts";
-import { assert, assertEquals, assertNotEOF } from "../testing/asserts.ts";
-import { Response, ServerRequest, serve } from "./server.ts";
+import {
+  assert,
+  assertEquals,
+  assertNotEOF,
+  assertStrContains
+} from "../testing/asserts.ts";
+import { Response, ServerRequest, Server, serve } from "./server.ts";
 import { BufReader, BufWriter } from "../io/bufio.ts";
 import { delay, deferred } from "../util/async.ts";
 import { encode, decode } from "../strings/mod.ts";
@@ -451,6 +456,38 @@ test("close server while iterating", async (): Promise<void> => {
   assertEquals(await nextAfterClosing, { value: undefined, done: true });
 });
 
+test({
+  name: "[http] close server while connection is open",
+  async fn(): Promise<void> {
+    async function iteratorReq(server: Server): Promise<void> {
+      for await (const req of server) {
+        req.respond({ body: new TextEncoder().encode(req.url) });
+      }
+    }
+
+    const server = serve(":8123");
+    iteratorReq(server);
+    const conn = await Deno.connect({ hostname: "127.0.0.1", port: 8123 });
+    await Deno.writeAll(
+      conn,
+      new TextEncoder().encode("GET /hello HTTP/1.1\r\n\r\n")
+    );
+    const res = new Uint8Array(100);
+    const nread = await conn.read(res);
+    assert(nread !== Deno.EOF);
+    const resStr = new TextDecoder().decode(res.subarray(0, nread));
+    assertStrContains(resStr, "/hello");
+    server.close();
+    // Defer to allow async ops to resolve after server has been closed.
+    await delay(0);
+    // Client connection should still be open, verify that
+    // it's visible in resource table.
+    const resources = Deno.resources();
+    assertEquals(resources[conn.rid], "tcpStream");
+    conn.close();
+  }
+});
+
 // TODO(kevinkassimo): create a test that works on Windows.
 // The following test is to ensure that if an error occurs during respond
 // would result in connection closed. (such that fd/resource is freed).