Refactor request listeners to use async functions

lib/cwmp.ts

@@ -59,6 +59,7 @@ import { parseXmlDeclaration } from "./xml-parser";
 import * as debug from "./debug";
 import { getRequestOrigin } from "./forwarded";
 import { getSocketEndpoints } from "./server";
+import { EventEmitter } from "events";
 
 const gzipPromisified = promisify(zlib.gzip);
 const deflatePromisified = promisify(zlib.deflate);
@@ -889,86 +890,98 @@ async function sendAcsRequest(
   return writeResponse(sessionContext, res);
 }
 
-// When socket closes, store active sessions in cache
-export function onConnection(socket: Socket): void {
-  socket.on("close", async () => {
-    const sessionContext = currentSessions.get(socket);
-    if (!sessionContext) return;
-    currentSessions.delete(socket);
-    if (sessionContext.authState !== 2) {
-      logger.accessError({
-        message: "Authentication failure",
-        sessionContext: sessionContext,
-      });
-      return;
-    }
+function once(
+  emitter: EventEmitter,
+  event: string,
+  timeout: number
+): Promise<unknown[]> {
+  return new Promise((resolve, reject) => {
+    const timer = setTimeout(() => {
+      reject(new Error(`Event ${event} timed out after ${timeout} ms`));
+    }, timeout);
 
-    const now = Date.now();
-
-    const lastActivity = sessionContext.lastActivity;
-    const timeoutMsg = {
-      sessionContext: sessionContext,
-      message: "Session timeout",
-      sessionTimestamp: sessionContext.timestamp,
-    };
-
-    const timeout =
-      sessionContext.lastActivity + sessionContext.timeout * 1000 - now;
-
-    if (timeout <= 0) {
-      logger.accessError(timeoutMsg);
-      // TODO it's possible that lock would have already been expired
-      await endSession(sessionContext);
-      return;
-    }
-
-    setTimeout(async () => {
-      const sessionContextString = await cache.get(
-        `session_${sessionContext.sessionId}`
-      );
-      if (!sessionContextString) return;
-      const _sessionContext = await session.deserialize(sessionContextString);
-      if (_sessionContext.lastActivity === lastActivity) {
-        logger.accessError(timeoutMsg);
-        await endSession(sessionContext);
-      }
-    }, timeout + 1000).unref();
-
-    if (sessionContext.state === 0) return;
-
-    const sessionContextString = await session.serialize(sessionContext);
-    await cache.set(
-      `session_${sessionContext.sessionId}`,
-      sessionContextString,
-      Math.ceil(timeout / 1000) + 3
-    );
+    emitter.once(event, (...args: unknown[]) => {
+      clearTimeout(timer);
+      resolve(args);
+    });
   });
 }
 
-export function onClientError(err: Error, socket: Socket): void {
-  const remoteAddress = getSocketEndpoints(socket).remoteAddress;
-  localCache
-    .getCurrentSnapshot()
-    .then((cacheSnapshot) => {
-      const debugEnabled = !!localCache.getConfig(
-        cacheSnapshot,
-        "cwmp.debug",
-        {
-          remoteAddress: remoteAddress,
-        },
-        Date.now(),
-        (e) => {
-          if (Array.isArray(e) && e[0] === "FUNC" && e[1] === "REMOTE_ADDRESS")
-            return remoteAddress;
-          return e;
-        }
-      );
+function setTimeoutPromise(delay: number, ref = true): Promise<void> {
+  return new Promise((resolve) => {
+    const timerId = setTimeout(resolve, delay);
+    if (!ref) timerId.unref();
+  });
+}
 
-      if (debugEnabled) debug.clientError(remoteAddress, err);
-    })
-    .catch((err) => {
-      throw err;
+// When socket closes, store active sessions in cache
+export async function onConnection(socket: Socket): Promise<void> {
+  await once(socket, "close", 300000);
+  const sessionContext = currentSessions.get(socket);
+  if (!sessionContext) return;
+  currentSessions.delete(socket);
+  if (sessionContext.authState !== 2) {
+    logger.accessError({
+      message: "Authentication failure",
+      sessionContext: sessionContext,
     });
+    return;
+  }
+
+  const now = Date.now();
+
+  const lastActivity = sessionContext.lastActivity;
+  const timeoutMsg = {
+    sessionContext: sessionContext,
+    message: "Session timeout",
+    sessionTimestamp: sessionContext.timestamp,
+  };
+
+  const timeout =
+    sessionContext.lastActivity + sessionContext.timeout * 1000 - now;
+
+  if (timeout <= 0) {
+    logger.accessError(timeoutMsg);
+    // TODO it's possible that lock would have already been expired
+    await endSession(sessionContext);
+    return;
+  }
+
+  await cache.set(
+    `session_${sessionContext.sessionId}`,
+    await session.serialize(sessionContext),
+    Math.ceil(timeout / 1000) + 3
+  );
+
+  await setTimeoutPromise(timeout + 1000, false);
+  const sessionStr = await cache.get(`session_${sessionContext.sessionId}`);
+  if (!sessionStr) return;
+
+  const _sessionContext = await session.deserialize(sessionStr);
+  if (_sessionContext.lastActivity === lastActivity) {
+    logger.accessError(timeoutMsg);
+    await endSession(sessionContext);
+  }
+}
+
+export async function onClientError(err: Error, socket: Socket): Promise<void> {
+  const remoteAddress = getSocketEndpoints(socket).remoteAddress;
+  const cacheSnapshot = await localCache.getCurrentSnapshot();
+  const debugEnabled = !!localCache.getConfig(
+    cacheSnapshot,
+    "cwmp.debug",
+    {
+      remoteAddress: remoteAddress,
+    },
+    Date.now(),
+    (e) => {
+      if (Array.isArray(e) && e[0] === "FUNC" && e[1] === "REMOTE_ADDRESS")
+        return remoteAddress;
+      return e;
+    }
+  );
+
+  if (debugEnabled) debug.clientError(remoteAddress, err);
 }
 
 setInterval(() => {
@@ -1341,29 +1354,19 @@ async function processRequest(
   }
 }
 
-export function listener(
+export async function listener(
   httpRequest: IncomingMessage,
   httpResponse: ServerResponse
-): void {
+): Promise<void> {
   stats.concurrentRequests += 1;
-  listenerAsync(httpRequest, httpResponse)
-    .then(() => {
-      stats.concurrentRequests -= 1;
-    })
-    .catch((err) => {
-      currentSessions.delete(httpRequest.socket);
-      stats.concurrentRequests -= 1;
-      setTimeout(() => {
-        throw err;
-      });
-      try {
-        httpRequest.socket.unref();
-        httpResponse.writeHead(500, { Connection: "close" });
-        httpResponse.end(`${err.name}: ${err.message}`);
-      } catch (err) {
-        // Ignore
-      }
-    });
+  try {
+    await listenerAsync(httpRequest, httpResponse);
+  } catch (err) {
+    currentSessions.delete(httpRequest.socket);
+    throw err;
+  } finally {
+    stats.concurrentRequests -= 1;
+  }
 }
 
 async function clientError(

lib/db.ts

@@ -89,7 +89,7 @@ onConnect(async (db) => {
 });
 
 export async function disconnect(): Promise<void> {
-  if (clientPromise) await (await clientPromise).close();
+  if (clientPromise != null) await (await clientPromise).close();
 }
 
 // Optimize projection by removing overlaps

lib/nbi.ts

@@ -66,10 +66,10 @@ async function getBody(request: IncomingMessage): Promise<Buffer> {
   return Buffer.concat(chunks);
 }
 
-export function listener(
+export async function listener(
   request: IncomingMessage,
   response: ServerResponse
-): void {
+): Promise<void> {
   response.setHeader("GenieACS-Version", VERSION);
 
   const origin = getRequestOrigin(request);
@@ -78,32 +78,17 @@ export function listener(
     (origin.encrypted ? "https://" : "http://") + origin.host
   );
 
-  let readableEnded = false;
-  // For Node 12.9+ we can just use stream.readableEnded
-  request.on("end", () => {
-    readableEnded = true;
-  });
+  const body = await getBody(request).catch(() => null);
+  // Ignore incomplete requests
+  if (body == null) return;
 
-  getBody(request)
-    .then((body) => {
-      logger.accessInfo(
-        Object.assign({}, Object.fromEntries(url.searchParams), {
-          remoteAddress: origin.remoteAddress,
-          message: `${request.method} ${url.pathname}`,
-        })
-      );
-      return handler(request, response, url, body);
+  logger.accessInfo(
+    Object.assign({}, Object.fromEntries(url.searchParams), {
+      remoteAddress: origin.remoteAddress,
+      message: `${request.method} ${url.pathname}`,
     })
-    .catch((err) => {
-      // Ignore incomplete requests
-      if (!readableEnded) return;
-
-      if (!response.headersSent) {
-        response.writeHead(500, { Connection: "close" });
-        response.end(`${err.name}: ${err.message}`);
-      }
-      throw err;
-    });
+  );
+  return handler(request, response, url, body);
 }
 
 async function handler(

lib/sandbox.ts

@@ -50,7 +50,7 @@ function runExtension(sessionContext, key, extCall): Promise<Fault> {
   }
 
   let prom = re.get(key);
-  if (!prom) {
+  if (prom == null) {
     re.set(
       key,
       (prom = new Promise((resolve, reject) => {

lib/server.ts

@@ -25,7 +25,8 @@ import * as path from "path";
 import { ROOT_DIR } from "./config";
 
 let server: http.Server | https.Server;
-let listener: (...args) => void;
+let listener: http.RequestListener;
+let stopping = false;
 
 function closeServer(timeout, callback): void {
   if (!server) return void callback();
@@ -73,11 +74,29 @@ interface SocketEndpoint {
 // Save this info as they're not accessible after a socket has been closed
 const socketEndpoints: WeakMap<Socket, SocketEndpoint> = new WeakMap();
 
+type Promisify<T extends (...args: any) => any> = (
+  ...args: Parameters<T>
+) => Promise<ReturnType<T>>;
+
 export function start(
   options: ServerOptions,
-  _listener: http.RequestListener
+  _listener: Promisify<http.RequestListener>
 ): void {
-  listener = _listener;
+  listener = (req, res) => {
+    if (stopping) res.setHeader("Connection", "close");
+    _listener(req, res).catch((err) => {
+      try {
+        res.socket.unref();
+        if (res.headersSent) {
+          res.writeHead(500, { Connection: "close" });
+          res.end(`${err.name}: ${err.message}`);
+        }
+      } catch (err) {
+        // Ignore
+      }
+      throw err;
+    });
+  };
 
   if (options.ssl) {
     const opts = {
@@ -124,7 +143,8 @@ export function start(
   server.listen({ port: options.port, host: options.host });
 }
 
-export function stop(): Promise<void> {
+export function stop(terminateConnections = true): Promise<void> {
+  stopping = terminateConnections;
   return new Promise((resolve, reject) => {
     setTimeout(() => {
       reject(new Error("Could not close server in a timely manner"));

lib/ui.ts

@@ -59,7 +59,7 @@ const getAuthorizer = memoize(
   }
 );
 
-koa.on("error", async (err, ctx) => {
+koa.on("error", (err, ctx) => {
   setTimeout(() => {
     // Ignored errors resulting from aborted requests
     if (ctx?.req.aborted) return;