Read session lock to detect task completion

lib/api-functions.ts

@@ -37,7 +37,7 @@ import { evaluate } from "./common/expression";
 export async function connectionRequest(
   deviceId: string,
   device?: Record<string, { value?: [boolean | number | string, string] }>
-): Promise<void> {
+): Promise<string> {
   if (!device) {
     const res = await db.devicesCollection.findOne({ _id: deviceId });
     if (!res) throw new Error("No such device");
@@ -156,8 +156,7 @@ export async function connectionRequest(
     );
   }
 
-  try {
-    await httpConnectionRequest(
+  const status = await httpConnectionRequest(
     connectionRequestUrl,
     authExp,
     CONNECTION_REQUEST_ALLOW_BASIC_AUTH,
@@ -165,34 +164,46 @@ export async function connectionRequest(
     debug,
     deviceId
   );
-  } catch (err) {
-    if (!udpProm) throw err;
+
+  if (udpProm) {
     await udpProm;
+    return "";
   }
+
+  return status;
 }
 
-export async function watchTask(
+export async function awaitSessionStart(
   deviceId: string,
-  taskId: string,
+  lastInform: number,
   timeout: number
-): Promise<string> {
-  await new Promise((resolve) => setTimeout(resolve, 500));
-
-  const task = await db.tasksCollection.findOne(
-    { _id: taskId },
-    { projection: { _id: 1 } }
+): Promise<boolean> {
+  const now = Date.now();
+  const device = await db.devicesCollection.findOne(
+    { _id: deviceId },
+    { projection: { _lastInform: 1 } }
   );
-  if (!task) return "completed";
+  const li = (device["_lastInform"] as Date).getTime();
+  if (li > lastInform) return true;
+  const token = await cache.get(`cwmp_session_${deviceId}`);
+  if (token) return true;
+  if (timeout < 500) return false;
+  await new Promise((resolve) => setTimeout(resolve, 500));
+  timeout -= Date.now() - now;
+  return awaitSessionStart(deviceId, lastInform, timeout);
+}
 
-  const q = { _id: `${deviceId}:task_${taskId}` };
-  const fault = await db.faultsCollection.findOne(q, {
-    projection: { _id: 1 },
-  });
-  if (fault) return "fault";
-
-  if ((timeout -= 500) <= 0) return "timeout";
-
-  return watchTask(deviceId, taskId, timeout);
+export async function awaitSessionEnd(
+  deviceId: string,
+  timeout: number
+): Promise<boolean> {
+  const now = Date.now();
+  const token = await cache.get(`cwmp_session_${deviceId}`);
+  if (!token) return true;
+  if (timeout < 500) return false;
+  await new Promise((resolve) => setTimeout(resolve, 500));
+  timeout -= Date.now() - now;
+  return awaitSessionEnd(deviceId, timeout);
 }
 
 function sanitizeTask(task): void {

lib/connection-request.ts

@@ -97,10 +97,10 @@ export async function httpConnectionRequest(
   timeout: number,
   _debug: boolean,
   deviceId: string
-): Promise<void> {
+): Promise<string> {
   const options: http.RequestOptions = parse(address);
   if (options.protocol !== "http:")
-    throw new Error("Invalid connection request URL or protocol");
+    return "Invalid connection request URL or protocol";
 
   options.agent = new http.Agent({
     maxSockets: 1,
@@ -115,8 +115,7 @@ export async function httpConnectionRequest(
     let opts = options;
     if (authHeader) {
       if (authHeader["method"] === "Basic") {
-        if (!allowBasicAuth)
-          throw new Error("Basic HTTP authentication not allowed");
+        if (!allowBasicAuth) return "Basic HTTP authentication not allowed";
 
         opts = Object.assign(
           {
@@ -143,7 +142,7 @@ export async function httpConnectionRequest(
           options
         );
       } else {
-        throw new Error("Unrecognized auth method");
+        return "Unrecognized auth method";
       }
     }
 
@@ -152,8 +151,8 @@ export async function httpConnectionRequest(
     // Workaround for some devices unexpectedly closing the connection
     if (res.statusCode === 0 && authHeader)
       res = await httpGet(opts, timeout, _debug, deviceId);
-    if (res.statusCode === 0) throw new Error("Device is offline");
-    if (res.statusCode === 200 || res.statusCode === 204) return;
+    if (res.statusCode === 0) return "Device is offline";
+    if (res.statusCode === 200 || res.statusCode === 204) return "";
 
     if (res.statusCode === 401 && res.headers["www-authenticate"]) {
       authHeader = auth.parseWwwAuthenticateHeader(
@@ -161,12 +160,10 @@ export async function httpConnectionRequest(
       );
       [username, password, authExp] = await extractAuth(authExp, false);
     } else {
-      throw new Error(
-        `Unexpected response code from device: ${res.statusCode}`
-      );
+      return `Unexpected response code from device: ${res.statusCode}`;
     }
   }
-  throw new Error("Incorrect connection request credentials");
+  return "Incorrect connection request credentials";
 }
 
 export async function udpConnectionRequest(

lib/nbi.ts

@@ -392,47 +392,91 @@ export function listener(
           task.device = deviceId;
           apiFunctions
             .insertTasks(task)
-            .then(() => {
-              cache
-                .del(`${deviceId}_tasks_faults_operations`)
-                .then(() => {
-                  if (urlParts.query.connection_request != null) {
-                    apiFunctions
-                      .connectionRequest(deviceId)
-                      .then(() => {
-                        const taskTimeout =
+            .then(async () => {
+              const lastInform = Date.now();
+              const notInSession = await apiFunctions.awaitSessionEnd(
+                deviceId,
+                30000
+              );
+              await cache.del(`${deviceId}_tasks_faults_operations`);
+              if (urlParts.query.connection_request == null) {
+                response.writeHead(202, {
+                  "Content-Type": "application/json",
+                });
+                response.end(JSON.stringify(task));
+                return;
+              }
+
+              if (!notInSession) {
+                response.writeHead(202, "Task queued but not processed", {
+                  "Content-Type": "application/json",
+                });
+                response.end(JSON.stringify(task));
+                return;
+              }
+
+              const status = await apiFunctions.connectionRequest(deviceId);
+
+              if (status) {
+                response.writeHead(202, status, {
+                  "Content-Type": "application/json",
+                });
+                response.end(JSON.stringify(task));
+                return;
+              }
+
+              const onlineThreshold =
                 (urlParts.query.timeout &&
                   parseInt(urlParts.query.timeout as string)) ||
-                          (config.get(
-                            "DEVICE_ONLINE_THRESHOLD",
-                            deviceId
-                          ) as number);
+                (config.get("DEVICE_ONLINE_THRESHOLD", deviceId) as number);
 
-                        apiFunctions
-                          .watchTask(deviceId, task._id, taskTimeout)
-                          .then((status) => {
-                            if (status === "timeout") {
-                              response.writeHead(
-                                202,
-                                "Task queued but not processed",
-                                {
+              const sessionStarted = await apiFunctions.awaitSessionStart(
+                deviceId,
+                lastInform,
+                onlineThreshold
+              );
+              if (!sessionStarted) {
+                response.writeHead(202, "Task queued but not processed", {
                   "Content-Type": "application/json",
+                });
+                response.end(JSON.stringify(task));
+                return;
+              }
+
+              const sessionEnded = await apiFunctions.awaitSessionEnd(
+                deviceId,
+                120000
+              );
+              if (!sessionEnded) {
+                response.writeHead(202, "Task queued but not processed", {
+                  "Content-Type": "application/json",
+                });
+                response.end(JSON.stringify(task));
+                return;
+              }
+
+              const prom1 = collections.tasks.findOne(
+                { _id: task._id },
+                { projection: { _id: 1 } }
+              );
+              const prom2 = collections.faults.findOne(
+                { _id: `${deviceId}:task_${task._id}` },
+                {
+                  projection: { _id: 1 },
                 }
               );
-                              response.end(JSON.stringify(task));
-                            } else if (status === "fault") {
-                              collections.tasks.findOne(
-                                { _id: task._id },
-                                (err, task2) => {
-                                  if (err)
-                                    return void throwError(err, response);
 
+              const [t, f] = await Promise.all([prom1, prom2]);
+              if (f) {
                 response.writeHead(202, "Task faulted", {
                   "Content-Type": "application/json",
                 });
-                                  response.end(JSON.stringify(task2));
-                                }
-                              );
+                response.end(JSON.stringify(t || task));
+              } else if (t) {
+                response.writeHead(202, "Task queued but not processed", {
+                  "Content-Type": "application/json",
+                });
+                response.end(JSON.stringify(t));
               } else {
                 response.writeHead(200, {
                   "Content-Type": "application/json",
@@ -445,31 +489,6 @@ export function listener(
                 throwError(err, response);
               });
             });
-                      })
-                      .catch((err) => {
-                        response.writeHead(202, err.message, {
-                          "Content-Type": "application/json",
-                        });
-                        response.end(JSON.stringify(task));
-                      });
-                  } else {
-                    response.writeHead(202, {
-                      "Content-Type": "application/json",
-                    });
-                    response.end(JSON.stringify(task));
-                  }
-                })
-                .catch((err) => {
-                  setTimeout(() => {
-                    throwError(err, response);
-                  });
-                });
-            })
-            .catch((err) => {
-              setTimeout(() => {
-                throwError(err, response);
-              });
-            });
         } else if (urlParts.query.connection_request != null) {
           // No task, send connection request only
           apiFunctions

lib/ui/api-functions.ts

@@ -24,7 +24,8 @@ import { getUsers } from "../local-cache";
 import { hashPassword } from "../auth";
 import {
   insertTasks,
-  watchTask,
+  awaitSessionEnd,
+  awaitSessionStart,
   connectionRequest,
   deleteDevice,
 } from "../api-functions";
@@ -104,23 +105,45 @@ export async function postTasks(
   const statuses = tasks.map((t) => {
     return { _id: t._id, status: "pending" };
   });
-
-  await del(`${deviceId}_tasks_faults_operations`);
-
-  try {
-    await connectionRequest(deviceId, device);
-  } catch (err) {
+  const lastInform = Date.now();
+  const notInSession = await awaitSessionEnd(deviceId, 30000);
+  if (!notInSession) {
     await Promise.all(statuses.map((t) => db.deleteTask(new ObjectID(t._id))));
     return {
-      connectionRequest: err.message,
+      connectionRequest: "Session took too long to complete",
       tasks: statuses,
     };
   }
 
-  const sample = tasks[tasks.length - 1];
+  await del(`${deviceId}_tasks_faults_operations`);
 
-  // Waiting for session to finish or timeout
-  await watchTask(deviceId, sample._id, timeout);
+  const status = await connectionRequest(deviceId, device);
+
+  if (status) {
+    await Promise.all(statuses.map((t) => db.deleteTask(new ObjectID(t._id))));
+    return {
+      connectionRequest: status,
+      tasks: statuses,
+    };
+  }
+
+  const sessionStarted = await awaitSessionStart(deviceId, lastInform, timeout);
+  if (!sessionStarted) {
+    await Promise.all(statuses.map((t) => db.deleteTask(new ObjectID(t._id))));
+    return {
+      connectionRequest: "No contact from CPE",
+      tasks: statuses,
+    };
+  }
+
+  const sessionEnded = await awaitSessionEnd(deviceId, 120000);
+  if (!sessionEnded) {
+    await Promise.all(statuses.map((t) => db.deleteTask(new ObjectID(t._id))));
+    return {
+      connectionRequest: "Session took too long to complete",
+      tasks: statuses,
+    };
+  }
 
   const promises = [];
   for (const s of statuses) {