Move locks to their own DB collection
This commit is contained in:
1 parent
c3954c277f
commit
02665faa9f
6 files changed
+61
-46
No files matched your search
@@ -52,45 +52,3 @@ export async function pop(key: string): Promise<string> {
|
|||||||
const res = await collections.cache.findOneAndDelete({ _id: key });
|
const res = await collections.cache.findOneAndDelete({ _id: key });
|
||||||
return res.value?.value;
|
return res.value?.value;
|
||||||
}
|
}
|
||||||
|
|
||||||
export async function acquireLock(
|
|
||||||
lockName: string,
|
|
||||||
ttl: number,
|
|
||||||
timeout = 0,
|
|
||||||
token = Math.random().toString(36).slice(2)
|
|
||||||
): Promise<string> {
|
|
||||||
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<void> {
|
|
||||||
const res = await collections.cache.deleteOne({
|
|
||||||
_id: lockName,
|
|
||||||
value: token,
|
|
||||||
});
|
|
||||||
if (res.deletedCount !== 1) throw new Error("Lock expired");
|
|
||||||
}
|
|
||||||
+4
-3
@@ -31,6 +31,7 @@ import {
|
|||||||
extractParams,
|
extractParams,
|
||||||
} from "./common/expression/util";
|
} from "./common/expression/util";
|
||||||
import * as cache from "./cache";
|
import * as cache from "./cache";
|
||||||
|
import * as lock from "./lock";
|
||||||
import * as localCache from "./cwmp/local-cache";
|
import * as localCache from "./cwmp/local-cache";
|
||||||
import {
|
import {
|
||||||
clearTasks,
|
clearTasks,
|
||||||
@@ -228,7 +229,7 @@ async function writeResponse(
|
|||||||
currentSessions.set(connection, sessionContext);
|
currentSessions.set(connection, sessionContext);
|
||||||
if (now >= sessionContext.extendLock) {
|
if (now >= sessionContext.extendLock) {
|
||||||
sessionContext.extendLock = now + 10000;
|
sessionContext.extendLock = now + 10000;
|
||||||
const lockToken = await cache.acquireLock(
|
const lockToken = await lock.acquireLock(
|
||||||
`cwmp_session_${sessionContext.deviceId}`,
|
`cwmp_session_${sessionContext.deviceId}`,
|
||||||
sessionContext.timeout * 1000 + 15000,
|
sessionContext.timeout * 1000 + 15000,
|
||||||
0,
|
0,
|
||||||
@@ -842,7 +843,7 @@ async function endSession(sessionContext: SessionContext): Promise<void> {
|
|||||||
}
|
}
|
||||||
|
|
||||||
await Promise.all(promises);
|
await Promise.all(promises);
|
||||||
await cache.releaseLock(
|
await lock.releaseLock(
|
||||||
`cwmp_session_${sessionContext.deviceId}`,
|
`cwmp_session_${sessionContext.deviceId}`,
|
||||||
sessionContext.sessionId
|
sessionContext.sessionId
|
||||||
);
|
);
|
||||||
@@ -1154,7 +1155,7 @@ async function processRequest(
|
|||||||
}
|
}
|
||||||
|
|
||||||
sessionContext.extendLock = sessionContext.timestamp + 10000;
|
sessionContext.extendLock = sessionContext.timestamp + 10000;
|
||||||
const lockToken = await cache.acquireLock(
|
const lockToken = await lock.acquireLock(
|
||||||
`cwmp_session_${sessionContext.deviceId}`,
|
`cwmp_session_${sessionContext.deviceId}`,
|
||||||
sessionContext.timeout * 1000 + 15000,
|
sessionContext.timeout * 1000 + 15000,
|
||||||
0,
|
0,
|
||||||
|
|||||||
@@ -37,6 +37,7 @@ export const collections = {
|
|||||||
users: null as Collection<MongoTypes.User>,
|
users: null as Collection<MongoTypes.User>,
|
||||||
config: null as Collection<MongoTypes.Config>,
|
config: null as Collection<MongoTypes.Config>,
|
||||||
cache: null as Collection<MongoTypes.Cache>,
|
cache: null as Collection<MongoTypes.Cache>,
|
||||||
|
locks: null as Collection<MongoTypes.Lock>,
|
||||||
};
|
};
|
||||||
|
|
||||||
let clientPromise: Promise<MongoClient>;
|
let clientPromise: Promise<MongoClient>;
|
||||||
@@ -60,11 +61,13 @@ export async function connect(): Promise<void> {
|
|||||||
collections.users = db.collection("users");
|
collections.users = db.collection("users");
|
||||||
collections.config = db.collection("config");
|
collections.config = db.collection("config");
|
||||||
collections.cache = db.collection("cache");
|
collections.cache = db.collection("cache");
|
||||||
|
collections.locks = db.collection("locks");
|
||||||
filesBucket = new GridFSBucket(db);
|
filesBucket = new GridFSBucket(db);
|
||||||
|
|
||||||
await Promise.all([
|
await Promise.all([
|
||||||
collections.tasks.createIndex({ device: 1, timestamp: 1 }),
|
collections.tasks.createIndex({ device: 1, timestamp: 1 }),
|
||||||
collections.cache.createIndex({ expire: 1 }, { expireAfterSeconds: 0 }),
|
collections.cache.createIndex({ expire: 1 }, { expireAfterSeconds: 0 }),
|
||||||
|
collections.locks.createIndex({ expire: 1 }, { expireAfterSeconds: 0 }),
|
||||||
]);
|
]);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -173,3 +173,10 @@ export interface User {
|
|||||||
roles: string;
|
roles: string;
|
||||||
salt: string;
|
salt: string;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
export interface Lock {
|
||||||
|
_id: string;
|
||||||
|
value: string;
|
||||||
|
timestamp: Date;
|
||||||
|
expire: Date;
|
||||||
|
}
|
||||||
+2
-1
@@ -18,7 +18,8 @@
|
|||||||
*/
|
*/
|
||||||
|
|
||||||
import { setTimeoutPromise } from "./util";
|
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 REFRESH = 5000;
|
||||||
const EVICT_TIMEOUT = 120000;
|
const EVICT_TIMEOUT = 120000;
|
||||||
|
|||||||
+45
@@ -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<string> {
|
||||||
|
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<void> {
|
||||||
|
const res = await collections.locks.deleteOne({
|
||||||
|
_id: lockName,
|
||||||
|
value: token,
|
||||||
|
});
|
||||||
|
if (res.deletedCount !== 1) throw new Error("Lock expired");
|
||||||
|
}
|
||||||
Reference in new issue
Block a user