Merge pull request #5498 from apollographql/glasser/inline-stoppable

Inline and improve the `stoppable` package

package.json

@@ -88,12 +88,12 @@
     "@types/qs-middleware": "1.0.1",
     "@types/request": "2.48.6",
     "@types/request-promise": "4.1.48",
-    "@types/stoppable": "1.1.1",
     "@types/supertest": "2.0.11",
     "@types/test-listen": "1.1.0",
     "@types/type-is": "1.6.3",
     "@types/uuid": "8.3.1",
     "@vendia/serverless-express": "4.3.9",
+    "awaiting": "^3.0.0",
     "azure-functions-ts-essentials": "1.3.2",
     "body-parser": "1.19.0",
     "bunyan": "1.8.15",
@@ -121,6 +121,7 @@
     "prettier": "2.3.2",
     "qs-middleware": "1.0.3",
     "request-promise": "4.2.6",
+    "requisition": "^1.7.0",
     "supertest": "6.1.4",
     "test-listen": "1.1.0",
     "ts-jest": "27.0.4",

packages/apollo-server/package.json

@@ -24,8 +24,7 @@
   "dependencies": {
     "apollo-server-core": "file:../apollo-server-core",
     "apollo-server-express": "file:../apollo-server-express",
-    "express": "^4.17.1",
-    "stoppable": "^1.1.0"
+    "express": "^4.17.1"
   },
   "devDependencies": {
     "apollo-server-integration-testsuite": "file:../apollo-server-integration-testsuite"

packages/apollo-server/src/__tests__/stoppable.test.ts

@@ -0,0 +1,277 @@
+// This file is adapted from the stoppable npm package:
+// https://github.com/hunterloftis/stoppable
+//
+// We've ported it to TypeScript and made some further changes.
+// Here's the license of the original code:
+//
+// The MIT License (MIT)
+//
+// Copyright (c) 2017 Hunter Loftis <hunter@hunterloftis.com>
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+import http from 'http';
+import https from 'https';
+const a: any = require('awaiting');
+const request: any = require('requisition');
+import fs from 'fs';
+import { Stopper } from '../stoppable';
+import child from 'child_process';
+import path from 'path';
+import { AddressInfo } from 'net';
+
+function port(s: http.Server) {
+  return (s.address() as AddressInfo).port;
+}
+
+interface SchemeInfo {
+  agent: (opts?: http.AgentOptions) => http.Agent;
+  server: (handler?: http.RequestListener) => http.Server;
+}
+
+const agents: http.Agent[] = [];
+afterEach(() => {
+  agents.forEach((a) => a.destroy());
+  agents.length = 0;
+});
+
+const schemes: Record<string, SchemeInfo> = {
+  http: {
+    agent: (opts = {}) => {
+      const a = new http.Agent(opts);
+      agents.push(a);
+      return a;
+    },
+    server: (handler) =>
+      http.createServer(handler || ((_req, res) => res.end('hello'))),
+  },
+  https: {
+    agent: (opts = {}) => {
+      const a = new https.Agent(
+        Object.assign({ rejectUnauthorized: false }, opts),
+      );
+      agents.push(a);
+      return a;
+    },
+    server: (handler) =>
+      https.createServer(
+        {
+          key: fs.readFileSync(
+            path.join(__dirname, 'stoppable', 'fixture.key'),
+          ),
+          cert: fs.readFileSync(
+            path.join(__dirname, 'stoppable', 'fixture.cert'),
+          ),
+        },
+        handler || ((_req, res) => res.end('hello')),
+      ),
+  },
+};
+
+Object.keys(schemes).forEach((schemeName) => {
+  const scheme = schemes[schemeName];
+
+  describe(`${schemeName}.Server`, function () {
+    describe('.close()', () => {
+      let server: http.Server;
+
+      beforeEach(function () {
+        server = scheme.server();
+      });
+
+      it('without keep-alive connections', async () => {
+        let closed = 0;
+        server.on('close', () => closed++);
+        server.listen(0);
+        const p = port(server);
+        await a.event(server, 'listening');
+        const res1 = await request(`${schemeName}://localhost:${p}`).agent(
+          scheme.agent(),
+        );
+        const text1 = await res1.text();
+        expect(text1).toBe('hello');
+        server.close();
+        const err = await a.failure(
+          request(`${schemeName}://localhost:${p}`).agent(scheme.agent()),
+        );
+        expect(err.message).toMatch(/ECONNREFUSED/);
+        expect(closed).toBe(1);
+      });
+
+      it('with keep-alive connections', async () => {
+        let closed = 0;
+
+        server.on('close', () => closed++);
+        server.listen(0);
+        const p = port(server);
+        await a.event(server, 'listening');
+        const res1 = await request(`${schemeName}://localhost:${p}`).agent(
+          scheme.agent({ keepAlive: true }),
+        );
+        const text1 = await res1.text();
+        expect(text1).toBe('hello');
+        server.close();
+        const err = await a.failure(
+          request(`${schemeName}://localhost:${p}`).agent(
+            scheme.agent({ keepAlive: true }),
+          ),
+        );
+        expect(err.message).toMatch(/ECONNREFUSED/);
+        expect(closed).toBe(0);
+      });
+    });
+
+    describe('Stopper', function () {
+      it('without keep-alive connections', async () => {
+        let closed = 0;
+        const server = scheme.server();
+        const stopper = new Stopper(server);
+
+        server.on('close', () => closed++);
+        server.listen(0);
+        const p = port(server);
+        await a.event(server, 'listening');
+        const res1 = await request(`${schemeName}://localhost:${p}`).agent(
+          scheme.agent(),
+        );
+        const text1 = await res1.text();
+        expect(text1).toBe('hello');
+        const gracefully = await stopper.stop();
+        const err = await a.failure(
+          request(`${schemeName}://localhost:${p}`).agent(scheme.agent()),
+        );
+        expect(err.message).toMatch(/ECONNREFUSED/);
+
+        expect(closed).toBe(1);
+        expect(gracefully).toBe(true);
+      });
+
+      it('with keep-alive connections', async () => {
+        let closed = 0;
+        const server = scheme.server();
+        const stopper = new Stopper(server);
+
+        server.on('close', () => closed++);
+        server.listen(0);
+        const p = port(server);
+        await a.event(server, 'listening');
+        const res1 = await request(`${schemeName}://localhost:${p}`).agent(
+          scheme.agent({ keepAlive: true }),
+        );
+        const text1 = await res1.text();
+        expect(text1).toBe('hello');
+        const gracefully = await stopper.stop();
+        const err = await a.failure(
+          request(`${schemeName}://localhost:${p}`).agent(
+            scheme.agent({ keepAlive: true }),
+          ),
+        );
+        expect(err.message).toMatch(/ECONNREFUSED/);
+
+        expect(closed).toBe(1);
+        expect(gracefully).toBe(true);
+
+        expect(stopper['reqsPerSocket'].size).toBe(0);
+      });
+    });
+
+    it('with a 0.5s grace period', async () => {
+      const server = scheme.server((_req, res) => {
+        res.writeHead(200);
+        res.write('hi');
+      });
+      const stopper = new Stopper(server);
+      server.listen(0);
+      const p = port(server);
+      await a.event(server, 'listening');
+      await Promise.all([
+        request(`${schemeName}://localhost:${p}`).agent(
+          scheme.agent({ keepAlive: true }),
+        ),
+        request(`${schemeName}://localhost:${p}`).agent(
+          scheme.agent({ keepAlive: true }),
+        ),
+      ]);
+      const start = Date.now();
+      const closeEventPromise = a.event(server, 'close');
+      const gracefully = await stopper.stop(500);
+      await closeEventPromise;
+      const elapsed = Date.now() - start;
+      expect(elapsed).toBeGreaterThanOrEqual(450);
+      expect(elapsed).toBeLessThanOrEqual(550);
+      expect(gracefully).toBe(false);
+      // It takes a moment for the `finish` events to happen.
+      await a.delay(20);
+      expect(stopper['reqsPerSocket'].size).toBe(0);
+    });
+
+    it('with requests in-flight', async () => {
+      const server = scheme.server((req, res) => {
+        const delay = parseInt(req.url!.slice(1), 10);
+        res.writeHead(200);
+        res.write('hello');
+        setTimeout(() => res.end('world'), delay);
+      });
+      const stopper = new Stopper(server);
+
+      server.listen(0);
+      const p = port(server);
+      await a.event(server, 'listening');
+      const start = Date.now();
+      const res = await Promise.all([
+        request(`${schemeName}://localhost:${p}/250`).agent(
+          scheme.agent({ keepAlive: true }),
+        ),
+        request(`${schemeName}://localhost:${p}/500`).agent(
+          scheme.agent({ keepAlive: true }),
+        ),
+      ]);
+      const closeEventPromise = a.event(server, 'close');
+      const gracefully = await stopper.stop();
+      const bodies = await Promise.all(res.map((r) => r.text()));
+      await closeEventPromise;
+      expect(bodies[0]).toBe('helloworld');
+      const elapsed = Date.now() - start;
+      expect(elapsed).toBeGreaterThanOrEqual(400);
+      expect(elapsed).toBeLessThanOrEqual(600);
+      expect(gracefully).toBe(true);
+    });
+
+    if (schemeName === 'http') {
+      it('with in-flights finishing before grace period ends', async () => {
+        const file = path.join(__dirname, 'stoppable', 'server.js');
+        const server = child.spawn('node', [file, '500']);
+        const [data] = await a.event(server.stdout, 'data');
+        const port = +data.toString();
+        expect(typeof port).toBe('number');
+        const start = Date.now();
+        const res = await request(
+          `${schemeName}://localhost:${port}/250`,
+        ).agent(scheme.agent({ keepAlive: true }));
+        const body = await res.text();
+        expect(body).toBe('helloworld');
+        const elapsed = Date.now() - start;
+        expect(elapsed).toBeGreaterThanOrEqual(150);
+        expect(elapsed).toBeLessThanOrEqual(350);
+        // Wait for subprocess to go away.
+        await a.event(server, 'close');
+      });
+    }
+  });
+});

packages/apollo-server/src/__tests__/stoppable/fixture.cert

@@ -0,0 +1,19 @@
+-----BEGIN CERTIFICATE-----
+MIIDBzCCAe+gAwIBAgIJALpsaWTYcddKMA0GCSqGSIb3DQEBBQUAMBoxGDAWBgNV
+BAMMD3d3dy5leGFtcGxlLmNvbTAeFw0xNzA1MTkxOTAwMzZaFw0yNzA1MTcxOTAw
+MzZaMBoxGDAWBgNVBAMMD3d3dy5leGFtcGxlLmNvbTCCASIwDQYJKoZIhvcNAQEB
+BQADggEPADCCAQoCggEBAOix4xb3jnIdWwqfT+vtoXadLuQJE21HN19Y2falkJuB
+k/ME0TKMeMjmQQZCO7G0K+5fMnQsEtQjcjY3XUXViliGfkAQplUD3Z9/xwsTSGN2
+ahBXE2W/GV+xJ6uX9KbJx2pMOXCuux6cKylYhmr8cTs2f9E6QpPji4LqtHv/9cAE
+QKRmv2rSAP1Q+1Ne2WYNbgHBuI35vuQsvZTN5QsozsferP9Qqtx8kpnBaLTgFZYD
+ZaEreYwFFYAQNfq2jOGEAAxStiXUpn3rT9T8KeOvLfWOifqYzDOTzL0t2py9bnvl
+x2fl8aJHc3NiU+4qlq3DuDEitiUoOkirGhFL7JFH4K0CAwEAAaNQME4wHQYDVR0O
+BBYEFAI/PRTwA3VKpSQAwXg2JDmDGVXxMB8GA1UdIwQYMBaAFAI/PRTwA3VKpSQA
+wXg2JDmDGVXxMAwGA1UdEwQFMAMBAf8wDQYJKoZIhvcNAQEFBQADggEBAHfYOl+n
+qB0Oyq/2jXE33fR5+QsaDrciXH0BWCjIVM8ADLPjYPAD3StTIceg3wfCw2n9xuku
+pObukGbApSqvGBSAIsvbbNAwcFp31DCqGQdfVMJMG1TCp3J7y0WslYMqU7DM1jYt
+ybqF9ICuGdPZ0KVsv0/ZmbSs98/fSyjUYoHfD+GngVrguuU/v0XUi4hjVqFyMZQZ
+AxGNq4QIlKxdo55L45vCMzGiajT7BE0EnChvFpOGXF5/pk072RESI7uxJBiAssWP
+uCk0xHxLtacOQK3seFFw0d7t3769gVDNi732eTMhoFQj+loSgmnRwDKL7QPhZ8tj
+pRRUGV4sPR+ucpo=
+-----END CERTIFICATE-----

packages/apollo-server/src/__tests__/stoppable/fixture.key

@@ -0,0 +1,27 @@
+-----BEGIN RSA PRIVATE KEY-----
+MIIEpAIBAAKCAQEA6LHjFveOch1bCp9P6+2hdp0u5AkTbUc3X1jZ9qWQm4GT8wTR
+Mox4yOZBBkI7sbQr7l8ydCwS1CNyNjddRdWKWIZ+QBCmVQPdn3/HCxNIY3ZqEFcT
+Zb8ZX7Enq5f0psnHakw5cK67HpwrKViGavxxOzZ/0TpCk+OLguq0e//1wARApGa/
+atIA/VD7U17ZZg1uAcG4jfm+5Cy9lM3lCyjOx96s/1Cq3HySmcFotOAVlgNloSt5
+jAUVgBA1+raM4YQADFK2JdSmfetP1Pwp468t9Y6J+pjMM5PMvS3anL1ue+XHZ+Xx
+okdzc2JT7iqWrcO4MSK2JSg6SKsaEUvskUfgrQIDAQABAoIBAE8UJRipCL+/OjFh
+8sc6+qRUxpq4euGoUikVCP3JRluSrbTo7i8/jcy4c2CtIZxCnqtjrsHMOJnfcfD6
+37fb2ig7jKw4/E3oAmkyA3LAGtmyZFkpPm5Vg0oB6nlmKr6D1EFLpjmlJ/I/IGvs
+qcGyCMkWvFlec0HPEpprKOr7EYkvQscy99ny7JswG1P9PELwo7tIeb0BioPYnFmF
+I0BPgI1lxDHKQTUPAao9rStiHsuPMCkw51qUgp/Z814ld4KDXCaWFQPy5riHDykH
+wm9n2hkM6pq4d6eHuMVj7CuBdp141k2BAnZdysMHpE9y1315+didoEcox8+zOLeO
+OC4XZAECgYEA/U98ld2YnVcSL9/Pr17SVG/q3iZaUueUkf+CEdj7LpvStedpky5W
+dOM7ad0kBcPqIafgn/O3teYjVl8FM0oOtOheMHHMkYxbXuECA5hkk7emu3oIJcAO
++9Pb/uGdufWmAVyRueRam4tubiLxv46xeGUmscCnwG78bj+rq74ATjcCgYEA6ypd
+qt/b43y4SHY4LDuuJk5jfC5MNXztIi3sOtuGoJNUlzC/hI/NNhEDhP3Pzo9c/i0k
+aCzyjhRyiaFK2SHQ5SQdCFi44PM+MptwFjY1KPGv20m5omfBgJOoF+Ad13qrUQF/
+b7/C5j3PZkOZfwaYO+erLeaayWKRJi2AEoXb9jsCgYEAnxAHuo/A4qQnXnqbDpNr
+Xew9Pqw0sbSLvbYFNjHbYKQmh2U+DVbeoV2DFHHxydEBN4sUaTyAUq+l5vmZ6WAK
+phz38FG1VHwfcA+41QsftQZwo274qMPWZNnfXkjMY1ZWnKpFM8aqAtxmRrCYv2Ha
+HTDfQGUqsZK/3ncK1LhltrcCgYBiJKk4wfpL42YpX6Ur2LBibj6Yud22SO/SXuYC
+3lE+PJ6GBqM3GKilEs6sNxz98Nj3fzF9hJyp7SCsDbNmEPXUW5D+RcDKqNlhV3uc
+2XywHMWuuAMQI0sfdQAnDrKFlj1fLkfYBGi7nDotTLMHz2HDRnkrS913hHpdO4oC
+sPjOtwKBgQDDiG7Vagk4SgPXt7zE0aSiFIIjJLpM28mES+k6zOtKAyOcTMHcDrI7
+YmSN1kq3w2g7RS5eMUpszqbUGoR6VDAjbgGAakDOno/uZWfEMjiQiKvRDSY1nmlc
+xSKubMZDf/OKUYTGasL1rqJJN7mxW2irptygc26NxMeAWZfgkmiPLg==
+-----END RSA PRIVATE KEY-----

packages/apollo-server/src/__tests__/stoppable/server.js

@@ -0,0 +1,14 @@
+const http = require('http');
+const { Stopper } = require('../../../dist/stoppable.js');
+
+const grace = Number(process.argv[2] || Infinity);
+let stopper;
+const server = http.createServer((req, res) => {
+  const delay = parseInt(req.url.slice(1), 10);
+  res.writeHead(200);
+  res.write('hello');
+  setTimeout(() => res.end('world'), delay);
+  stopper.stop(grace);
+});
+stopper = new Stopper(server);
+server.listen(0, () => console.log(server.address().port));