From a22d7d4b10763d060315c6bb94f3186332fccb5c Mon Sep 17 00:00:00 2001 From: TheBeastLT Date: Mon, 22 Jan 2024 21:58:53 +0200 Subject: [PATCH 1/8] improve psql pool config --- addon/lib/cache.js | 2 +- addon/lib/repository.js | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/addon/lib/cache.js b/addon/lib/cache.js index 48a3d35..7f46d0a 100644 --- a/addon/lib/cache.js +++ b/addon/lib/cache.js @@ -7,7 +7,7 @@ const STREAM_KEY_PREFIX = `${GLOBAL_KEY_PREFIX}|stream`; const AVAILABILITY_KEY_PREFIX = `${GLOBAL_KEY_PREFIX}|availability`; const RESOLVED_URL_KEY_PREFIX = `${GLOBAL_KEY_PREFIX}|resolved`; -const STREAM_TTL = process.env.STREAM_TTL || 4 * 60 * 60; // 4 hours +const STREAM_TTL = process.env.STREAM_TTL || 24 * 60 * 60; // 24 hours const STREAM_EMPTY_TTL = process.env.STREAM_EMPTY_TTL || 60; // 1 minute const AVAILABILITY_TTL = 8 * 60 * 60; // 8 hours const AVAILABILITY_EMPTY_TTL = 30 * 60; // 30 minutes diff --git a/addon/lib/repository.js b/addon/lib/repository.js index cdee1ac..9447514 100644 --- a/addon/lib/repository.js +++ b/addon/lib/repository.js @@ -3,7 +3,7 @@ const Op = Sequelize.Op; const DATABASE_URI = process.env.DATABASE_URI; -const database = new Sequelize(DATABASE_URI, { logging: false, pool: { max: 50 } }); +const database = new Sequelize(DATABASE_URI, { logging: false, pool: { max: 50, min: 5, idle: 60 * 60 * 1000 } }); const Torrent = database.define('torrent', { From a71a846dbd2bf264ca7e620279d4a5f658ecf834 Mon Sep 17 00:00:00 2001 From: TheBeastLT Date: Fri, 26 Jan 2024 00:09:18 +0200 Subject: [PATCH 2/8] bring back sw stats --- addon/index.js | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/addon/index.js b/addon/index.js index 73606ab..ab0f0a4 100644 --- a/addon/index.js +++ b/addon/index.js @@ -1,9 +1,22 @@ import express from 'express'; +import swStats from 'swagger-stats'; import serverless from './serverless.js'; +import { manifest } from './lib/manifest.js'; import { initBestTrackers } from './lib/magnetHelper.js'; const app = express(); app.enable('trust proxy'); +app.use(swStats.getMiddleware({ + name: manifest().name, + version: manifest().version, + timelineBucketDuration: 60 * 60 * 1000, + apdexThreshold: 100, + authentication: true, + onAuthenticate: (req, username, password) => { + return username === process.env.METRICS_USER + && password === process.env.METRICS_PASSWORD + }, +})) app.use(express.static('static', { maxAge: '1y' })); app.use((req, res, next) => serverless(req, res, next)); app.listen(process.env.PORT || 7000, () => { From 3f1624cd8ec2d5e098a234512ff779718fc25c1f Mon Sep 17 00:00:00 2001 From: TheBeastLT Date: Fri, 26 Jan 2024 00:14:52 +0200 Subject: [PATCH 3/8] add temp logging --- addon/addon.js | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/addon/addon.js b/addon/addon.js index f7566e0..d40f243 100644 --- a/addon/addon.js +++ b/addon/addon.js @@ -70,10 +70,18 @@ builder.defineMetaHandler((args) => { }) async function streamHandler(args) { + console.log(`Current stats: `, limiter.counts()) + const start = Date.now(); if (args.type === Type.MOVIE) { - return movieRecordsHandler(args); + return movieRecordsHandler(args).then(result => { + console.log(`Execution time: ${Date.now() - start} ms`); + return result; + }); } else if (args.type === Type.SERIES) { - return seriesRecordsHandler(args); + return seriesRecordsHandler(args).then(result => { + console.log(`Execution time: ${Date.now() - start} ms`); + return result; + }); } return Promise.reject('not supported type'); } From 0a06018f83267286d5f1b9e5c5236a99dbc3a8e3 Mon Sep 17 00:00:00 2001 From: TheBeastLT Date: Fri, 26 Jan 2024 00:27:17 +0200 Subject: [PATCH 4/8] reduce limiter concurrent --- addon/addon.js | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/addon/addon.js b/addon/addon.js index d40f243..b3ab5a8 100644 --- a/addon/addon.js +++ b/addon/addon.js @@ -18,8 +18,8 @@ const STALE_ERROR_AGE = 7 * 24 * 60 * 60; // 7 days const builder = new addonBuilder(dummyManifest()); const limiter = new Bottleneck({ - maxConcurrent: process.env.LIMIT_MAX_CONCURRENT || 100, - highWater: process.env.LIMIT_QUEUE_SIZE || 120, + maxConcurrent: process.env.LIMIT_MAX_CONCURRENT || 40, + highWater: process.env.LIMIT_QUEUE_SIZE || 60, strategy: Bottleneck.strategy.OVERFLOW }); const limiterOptions = { expiration: 2 * 60 * 1000 } From af7511a6101b839f82323b3d0a44dc43e36ee44a Mon Sep 17 00:00:00 2001 From: TheBeastLT Date: Fri, 26 Jan 2024 01:12:02 +0200 Subject: [PATCH 5/8] export named queue creation --- addon/lib/namedQueue.js | 11 +++++++++++ addon/moch/moch.js | 10 +++------- 2 files changed, 14 insertions(+), 7 deletions(-) create mode 100644 addon/lib/namedQueue.js diff --git a/addon/lib/namedQueue.js b/addon/lib/namedQueue.js new file mode 100644 index 0000000..695ce05 --- /dev/null +++ b/addon/lib/namedQueue.js @@ -0,0 +1,11 @@ +import namedQueue from "named-queue"; + +export function createNamedQueue(concurrency) { + const queue = new namedQueue((task, callback) => task.method() + .then(result => callback(false, result)) + .catch((error => callback(error))), 200); + queue.wrap = (id, method) => new Promise(((resolve, reject) => { + queue.push({ id, method }, (error, result) => result ? resolve(result) : reject(error)); + })); + return queue; +} \ No newline at end of file diff --git a/addon/moch/moch.js b/addon/moch/moch.js index 0dfb1e4..82376bf 100644 --- a/addon/moch/moch.js +++ b/addon/moch/moch.js @@ -10,6 +10,7 @@ import StaticResponse, { isStaticUrl } from './static.js'; import { cacheWrapResolvedUrl } from '../lib/cache.js'; import { timeout } from '../lib/promises.js'; import { BadTokenError, streamFilename, AccessDeniedError, enrichMeta } from './mochHelper.js'; +import { createNamedQueue } from "../lib/namedQueue.js"; const RESOLVE_TIMEOUT = 2 * 60 * 1000; // 2 minutes const MIN_API_KEY_SYMBOLS = 15; @@ -62,9 +63,7 @@ export const MochOptions = { const unrestrictQueues = {} Object.values(MochOptions) .map(moch => moch.key) - .forEach(mochKey => unrestrictQueues[mochKey] = new namedQueue((task, callback) => task.method() - .then(result => callback(false, result)) - .catch((error => callback(error))), 200)); + .forEach(mochKey => unrestrictQueues[mochKey] = createNamedQueue(50)); export function hasMochConfigured(config) { return Object.keys(MochOptions).find(moch => config?.[moch]) @@ -110,10 +109,7 @@ export async function resolve(parameters) { return StaticResponse.FAILED_UNEXPECTED; }) .then(url => isStaticUrl(url) ? `${parameters.host}/${url}` : url); - const unrestrictQueue = unrestrictQueues[moch.key]; - return new Promise(((resolve, reject) => { - unrestrictQueue.push({ id, method }, (error, result) => result ? resolve(result) : reject(error)); - })); + return unrestrictQueues[moch.key].wrap(id, method); } export async function getMochCatalog(mochKey, config) { From 4681cb1f1bdc1e1954922759cced632bd01f801d Mon Sep 17 00:00:00 2001 From: TheBeastLT Date: Fri, 26 Jan 2024 01:12:44 +0200 Subject: [PATCH 6/8] add named queue for stream requests --- addon/addon.js | 27 +++++++++++++-------------- 1 file changed, 13 insertions(+), 14 deletions(-) diff --git a/addon/addon.js b/addon/addon.js index b3ab5a8..84bd488 100644 --- a/addon/addon.js +++ b/addon/addon.js @@ -9,6 +9,7 @@ import applySorting from './lib/sort.js'; import applyFilters from './lib/filter.js'; import { applyMochs, getMochCatalog, getMochItemMeta } from './moch/moch.js'; import StaticLinks from './moch/static.js'; +import { createNamedQueue } from "./lib/namedQueue.js"; const CACHE_MAX_AGE = parseInt(process.env.CACHE_MAX_AGE) || 60 * 60; // 1 hour in seconds const CACHE_MAX_AGE_EMPTY = 60; // 60 seconds @@ -17,9 +18,10 @@ const STALE_REVALIDATE_AGE = 4 * 60 * 60; // 4 hours const STALE_ERROR_AGE = 7 * 24 * 60 * 60; // 7 days const builder = new addonBuilder(dummyManifest()); +const requestQueue = createNamedQueue(200); const limiter = new Bottleneck({ maxConcurrent: process.env.LIMIT_MAX_CONCURRENT || 40, - highWater: process.env.LIMIT_QUEUE_SIZE || 60, + highWater: process.env.LIMIT_QUEUE_SIZE || 100, strategy: Bottleneck.strategy.OVERFLOW }); const limiterOptions = { expiration: 2 * 60 * 1000 } @@ -29,10 +31,7 @@ builder.defineStreamHandler((args) => { return Promise.resolve({ streams: [] }); } - return cacheWrapStream(args.id, () => limiter.schedule(limiterOptions, () => streamHandler(args) - .then(records => records - .sort((a, b) => b.torrent.seeders - a.torrent.seeders || b.torrent.uploadDate - a.torrent.uploadDate) - .map(record => toStreamInfo(record))))) + return requestQueue.wrap(args, () => resolveStreams(args)) .then(streams => applyFilters(streams, args.extra)) .then(streams => applySorting(streams, args.extra, args.type)) .then(streams => applyStaticInfo(streams)) @@ -69,19 +68,19 @@ builder.defineMetaHandler((args) => { }); }) +async function resolveStreams(args) { + return cacheWrapStream(args.id, () => limiter.schedule(limiterOptions, () => streamHandler(args) + .then(records => records + .sort((a, b) => b.torrent.seeders - a.torrent.seeders || b.torrent.uploadDate - a.torrent.uploadDate) + .map(record => toStreamInfo(record))))); +} + async function streamHandler(args) { console.log(`Current stats: `, limiter.counts()) - const start = Date.now(); if (args.type === Type.MOVIE) { - return movieRecordsHandler(args).then(result => { - console.log(`Execution time: ${Date.now() - start} ms`); - return result; - }); + return movieRecordsHandler(args); } else if (args.type === Type.SERIES) { - return seriesRecordsHandler(args).then(result => { - console.log(`Execution time: ${Date.now() - start} ms`); - return result; - }); + return seriesRecordsHandler(args); } return Promise.reject('not supported type'); } From 9366ad1ccb09d436058f3f8d9333ac29d4deedc6 Mon Sep 17 00:00:00 2001 From: TheBeastLT Date: Fri, 26 Jan 2024 01:20:35 +0200 Subject: [PATCH 7/8] fix request named queue --- addon/addon.js | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/addon/addon.js b/addon/addon.js index 84bd488..decc56b 100644 --- a/addon/addon.js +++ b/addon/addon.js @@ -31,7 +31,7 @@ builder.defineStreamHandler((args) => { return Promise.resolve({ streams: [] }); } - return requestQueue.wrap(args, () => resolveStreams(args)) + return requestQueue.wrap(args.id, () => resolveStreams(args)) .then(streams => applyFilters(streams, args.extra)) .then(streams => applySorting(streams, args.extra, args.type)) .then(streams => applyStaticInfo(streams)) From 64b8a55f3b5e78de10eb73ef6b1dcf72357c8fee Mon Sep 17 00:00:00 2001 From: TheBeastLT Date: Fri, 26 Jan 2024 01:30:25 +0200 Subject: [PATCH 8/8] do not limit stream request named queue --- addon/addon.js | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/addon/addon.js b/addon/addon.js index decc56b..a017a4e 100644 --- a/addon/addon.js +++ b/addon/addon.js @@ -18,7 +18,7 @@ const STALE_REVALIDATE_AGE = 4 * 60 * 60; // 4 hours const STALE_ERROR_AGE = 7 * 24 * 60 * 60; // 7 days const builder = new addonBuilder(dummyManifest()); -const requestQueue = createNamedQueue(200); +const requestQueue = createNamedQueue(Infinity); const limiter = new Bottleneck({ maxConcurrent: process.env.LIMIT_MAX_CONCURRENT || 40, highWater: process.env.LIMIT_QUEUE_SIZE || 100,