diff --git a/lib/cache.ts b/lib/cache.ts index 0dfe5bb..161b0b6 100644 --- a/lib/cache.ts +++ b/lib/cache.ts @@ -52,45 +52,3 @@ export async function pop(key: string): Promise { const res = await collections.cache.findOneAndDelete({ _id: key }); return res.value?.value; } - -export async function acquireLock( - lockName: string, - ttl: number, - timeout = 0, - token = Math.random().toString(36).slice(2) -): Promise { - try { - const now = Date.now(); - const r = await collections.cache.findOneAndUpdate( - { _id: lockName, value: token }, - { - $set: { - expire: new Date(now + ttl + CLOCK_SKEW_TOLERANCE), - }, - $currentDate: { timestamp: true }, - }, - { upsert: true, returnDocument: "after" } - ); - if (Math.abs(r.value.timestamp.getTime() - now) > CLOCK_SKEW_TOLERANCE) - throw new Error("Database clock skew too great"); - } catch (err) { - if (err.code !== 11000) throw err; - if (!(timeout > 0)) return null; - const w = 50 + Math.random() * 50; - await new Promise((resolve) => setTimeout(resolve, w)); - return acquireLock(lockName, ttl, timeout - w, token); - } - - return token; -} - -export async function releaseLock( - lockName: string, - token: string -): Promise { - const res = await collections.cache.deleteOne({ - _id: lockName, - value: token, - }); - if (res.deletedCount !== 1) throw new Error("Lock expired"); -} diff --git a/lib/cwmp.ts b/lib/cwmp.ts index bb4a45a..c601508 100644 --- a/lib/cwmp.ts +++ b/lib/cwmp.ts @@ -31,6 +31,7 @@ import { extractParams, } from "./common/expression/util"; import * as cache from "./cache"; +import * as lock from "./lock"; import * as localCache from "./cwmp/local-cache"; import { clearTasks, @@ -228,7 +229,7 @@ async function writeResponse( currentSessions.set(connection, sessionContext); if (now >= sessionContext.extendLock) { sessionContext.extendLock = now + 10000; - const lockToken = await cache.acquireLock( + const lockToken = await lock.acquireLock( `cwmp_session_${sessionContext.deviceId}`, sessionContext.timeout * 1000 + 15000, 0, @@ -842,7 +843,7 @@ async function endSession(sessionContext: SessionContext): Promise { } await Promise.all(promises); - await cache.releaseLock( + await lock.releaseLock( `cwmp_session_${sessionContext.deviceId}`, sessionContext.sessionId ); @@ -1154,7 +1155,7 @@ async function processRequest( } sessionContext.extendLock = sessionContext.timestamp + 10000; - const lockToken = await cache.acquireLock( + const lockToken = await lock.acquireLock( `cwmp_session_${sessionContext.deviceId}`, sessionContext.timeout * 1000 + 15000, 0, diff --git a/lib/db/db.ts b/lib/db/db.ts index d9965ad..4708608 100644 --- a/lib/db/db.ts +++ b/lib/db/db.ts @@ -37,6 +37,7 @@ export const collections = { users: null as Collection, config: null as Collection, cache: null as Collection, + locks: null as Collection, }; let clientPromise: Promise; @@ -60,11 +61,13 @@ export async function connect(): Promise { collections.users = db.collection("users"); collections.config = db.collection("config"); collections.cache = db.collection("cache"); + collections.locks = db.collection("locks"); filesBucket = new GridFSBucket(db); await Promise.all([ collections.tasks.createIndex({ device: 1, timestamp: 1 }), collections.cache.createIndex({ expire: 1 }, { expireAfterSeconds: 0 }), + collections.locks.createIndex({ expire: 1 }, { expireAfterSeconds: 0 }), ]); } diff --git a/lib/db/types.ts b/lib/db/types.ts index fc8ae04..e268fed 100644 --- a/lib/db/types.ts +++ b/lib/db/types.ts @@ -173,3 +173,10 @@ export interface User { roles: string; salt: string; } + +export interface Lock { + _id: string; + value: string; + timestamp: Date; + expire: Date; +} diff --git a/lib/local-cache.ts b/lib/local-cache.ts index 1bb67da..3b6bb59 100644 --- a/lib/local-cache.ts +++ b/lib/local-cache.ts @@ -18,7 +18,8 @@ */ import { setTimeoutPromise } from "./util"; -import { acquireLock, get, releaseLock, set } from "./cache"; +import { get, set } from "./cache"; +import { acquireLock, releaseLock } from "./lock"; const REFRESH = 5000; const EVICT_TIMEOUT = 120000; diff --git a/lib/lock.ts b/lib/lock.ts new file mode 100644 index 0000000..d2eac0e --- /dev/null +++ b/lib/lock.ts @@ -0,0 +1,45 @@ +import { collections } from "./db/db"; + +const CLOCK_SKEW_TOLERANCE = 30000; + +export async function acquireLock( + lockName: string, + ttl: number, + timeout = 0, + token = Math.random().toString(36).slice(2) +): Promise { + try { + const now = Date.now(); + const r = await collections.locks.findOneAndUpdate( + { _id: lockName, value: token }, + { + $set: { + expire: new Date(now + ttl + CLOCK_SKEW_TOLERANCE), + }, + $currentDate: { timestamp: true }, + }, + { upsert: true, returnDocument: "after" } + ); + if (Math.abs(r.value.timestamp.getTime() - now) > CLOCK_SKEW_TOLERANCE) + throw new Error("Database clock skew too great"); + } catch (err) { + if (err.code !== 11000) throw err; + if (!(timeout > 0)) return null; + const w = 50 + Math.random() * 50; + await new Promise((resolve) => setTimeout(resolve, w)); + return acquireLock(lockName, ttl, timeout - w, token); + } + + return token; +} + +export async function releaseLock( + lockName: string, + token: string +): Promise { + const res = await collections.locks.deleteOne({ + _id: lockName, + value: token, + }); + if (res.deletedCount !== 1) throw new Error("Lock expired"); +}