From fa1e40e37076a4f2b3abc4ddd228f1bb7360c526 Mon Sep 17 00:00:00 2001 From: Zaid Abdulla Date: Mon, 30 Nov 2020 18:11:51 -0800 Subject: [PATCH] Read session lock to detect task completion --- lib/api-functions.ts | 71 ++++++++++------- lib/connection-request.ts | 19 ++--- lib/nbi.ts | 161 +++++++++++++++++++++----------------- lib/ui/api-functions.ts | 45 ++++++++--- 4 files changed, 173 insertions(+), 123 deletions(-) diff --git a/lib/api-functions.ts b/lib/api-functions.ts index 1ad83dd..df8a414 100644 --- a/lib/api-functions.ts +++ b/lib/api-functions.ts @@ -37,7 +37,7 @@ import { evaluate } from "./common/expression"; export async function connectionRequest( deviceId: string, device?: Record -): Promise { +): Promise { if (!device) { const res = await db.devicesCollection.findOne({ _id: deviceId }); if (!res) throw new Error("No such device"); @@ -156,43 +156,54 @@ export async function connectionRequest( ); } - try { - await httpConnectionRequest( - connectionRequestUrl, - authExp, - CONNECTION_REQUEST_ALLOW_BASIC_AUTH, - CONNECTION_REQUEST_TIMEOUT, - debug, - deviceId - ); - } catch (err) { - if (!udpProm) throw err; + const status = await httpConnectionRequest( + connectionRequestUrl, + authExp, + CONNECTION_REQUEST_ALLOW_BASIC_AUTH, + CONNECTION_REQUEST_TIMEOUT, + debug, + deviceId + ); + + if (udpProm) { await udpProm; + return ""; } + + return status; } -export async function watchTask( +export async function awaitSessionStart( deviceId: string, - taskId: string, + lastInform: number, timeout: number -): Promise { - await new Promise((resolve) => setTimeout(resolve, 500)); - - const task = await db.tasksCollection.findOne( - { _id: taskId }, - { projection: { _id: 1 } } +): Promise { + 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 { + 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 { diff --git a/lib/connection-request.ts b/lib/connection-request.ts index cd317e2..fb9d512 100644 --- a/lib/connection-request.ts +++ b/lib/connection-request.ts @@ -97,10 +97,10 @@ export async function httpConnectionRequest( timeout: number, _debug: boolean, deviceId: string -): Promise { +): Promise { 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( diff --git a/lib/nbi.ts b/lib/nbi.ts index 11188da..e696158 100644 --- a/lib/nbi.ts +++ b/lib/nbi.ts @@ -392,78 +392,97 @@ 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 = - (urlParts.query.timeout && - parseInt(urlParts.query.timeout as string)) || - (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", - { - "Content-Type": "application/json", - } - ); - response.end(JSON.stringify(task)); - } else if (status === "fault") { - collections.tasks.findOne( - { _id: task._id }, - (err, task2) => { - if (err) - return void throwError(err, response); - - response.writeHead(202, "Task faulted", { - "Content-Type": "application/json", - }); - response.end(JSON.stringify(task2)); - } - ); - } else { - response.writeHead(200, { - "Content-Type": "application/json", - }); - response.end(JSON.stringify(task)); - } - }) - .catch((err) => { - setTimeout(() => { - 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); - }); + .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); + + 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 }, + } + ); + + const [t, f] = await Promise.all([prom1, prom2]); + if (f) { + response.writeHead(202, "Task faulted", { + "Content-Type": "application/json", + }); + 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", + }); + response.end(JSON.stringify(task)); + } }) .catch((err) => { setTimeout(() => { diff --git a/lib/ui/api-functions.ts b/lib/ui/api-functions.ts index 60d6b5b..b382a45 100644 --- a/lib/ui/api-functions.ts +++ b/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) {