diff --git a/controller/emby.js b/controller/emby.js index 9206cb2..f665940 100644 --- a/controller/emby.js +++ b/controller/emby.js @@ -1,18 +1,13 @@ 'use strict'; const conf = require('>/conf'); +const { fetchWithRetry } = require('./retry'); // Ask Emby to (re)scan its libraries so a newly filed item gets indexed. async function refresh(){ - let res = await fetch(`${conf.emby.url}/Library/Refresh?api_key=${conf.emby.apiKey}`, { + let res = await fetchWithRetry(`${conf.emby.url}/Library/Refresh?api_key=${conf.emby.apiKey}`, { method: 'POST', - }); - if(!res.ok){ - let error = new Error('EmbyError'); - error.message = `Emby library refresh failed( ${res.status})`; - error.status = 502; - throw error; - } + }, { label: 'emby refresh' }); return true; } diff --git a/controller/library.js b/controller/library.js index cd45dd3..7746518 100644 --- a/controller/library.js +++ b/controller/library.js @@ -1,6 +1,7 @@ 'use strict'; const conf = require('>/conf'); +const { fetchJSONWithRetry } = require('./retry'); // Resolution rank so we can compare "what you own" against a candidate release. function rankFromWidth(w){ @@ -25,9 +26,8 @@ function labelFromRank(rank){ return ['', '480p', '720p', '1080p', '4k'][rank] | async function embyGet(pathAndQuery){ let sep = pathAndQuery.includes('?') ? '&' : '?'; - let res = await fetch(`${conf.emby.url}${pathAndQuery}${sep}api_key=${conf.emby.apiKey}`); - if(!res.ok) throw new Error(`Emby ${res.status}`); - return await res.json(); + let res = await fetchJSONWithRetry(`${conf.emby.url}${pathAndQuery}${sep}api_key=${conf.emby.apiKey}`, {}, { label: `emby ${pathAndQuery.split('?')[0]}` }); + return res; } // Is this movie already in the library, and at what best quality? Never throws — an diff --git a/controller/ollama.js b/controller/ollama.js index 10514b8..3e7359e 100644 --- a/controller/ollama.js +++ b/controller/ollama.js @@ -41,7 +41,10 @@ async function chatJSON(system, user){ { role: 'user', content: user }, ], }); - return extractJSON(res.message && res.message.content); + // The ollama package may return message.content as a JSON string when format:'json' is used. + let content = res.message && res.message.content; + if(typeof content === 'string') return extractJSON(content); + return content; } module.exports = { client, chatJSON, extractJSON }; diff --git a/controller/organize.js b/controller/organize.js index 4e350aa..493f225 100644 --- a/controller/organize.js +++ b/controller/organize.js @@ -7,6 +7,7 @@ const ollama = require('>/controller/ollama'); const tmdb = require('>/controller/tmdb'); const emby = require('>/controller/emby'); const ps = require('>/controller/pubsub'); +const { withRetry } = require('>/controller/retry'); const VIDEO_EXT = new Set(['.mkv', '.mp4', '.avi', '.m4v', '.ts', '.wmv', '.mov']); const SUB_EXT = new Set(['.srt', '.ass', '.ssa', '.sub', '.vtt', '.idx']); @@ -57,7 +58,10 @@ function buildExtractPrompt(today){ async function llmExtract({ name, category, files }){ let today = new Date().toISOString().slice(0, 10); let user = JSON.stringify({ name, type: category, files }); - return await ollama.chatJSON(buildExtractPrompt(today), user); + return await withRetry( + () => ollama.chatJSON(buildExtractPrompt(today), user), + { label: 'ollama llmExtract' } + ); } // Parse + canonicalize. Returns a metadata object, or one carrying `organizeError` diff --git a/controller/retry.js b/controller/retry.js new file mode 100644 index 0000000..24d0c25 --- /dev/null +++ b/controller/retry.js @@ -0,0 +1,85 @@ +'use strict'; + +// Retry an async function with exponential backoff up to a max delay. +// No attempt limit: keeps retrying on transient failures forever. +// Transient = network errors, 5xx, 429. 4xx client errors are NOT retried by default. +const TRANSIENT_NETWORK_CODES = new Set([ + 'ECONNREFUSED', 'ECONNRESET', 'ETIMEDOUT', 'EAI_AGAIN', + 'ENOTFOUND', 'EPIPE', 'ERR_SOCKET_TIMEOUT', 'ETIMEOUT', +]); + +function isRetryable(error){ + // Network / DNS / socket errors + if(error && error.code && TRANSIENT_NETWORK_CODES.has(error.code)) return true; + + // HTTP status codes worth retrying + let status = Number(error && error.status) || Number(error && error.status_code); + if(status >= 500 && status < 600) return true; + if(status === 429) return true; + if(status === 502 || status === 503 || status === 504) return true; + + // ResponseError from the ollama package carries status_code + if(error && error.name === 'ResponseError' && status >= 500) return true; + + return false; +} + +async function sleep(ms){ + return new Promise(resolve => setTimeout(resolve, ms)); +} + +async function withRetry(fn, options = {}){ + let baseDelay = options.baseDelay || 1000; + let maxDelay = options.maxDelay || 30000; + let multiplier = options.multiplier || 2; + let jitter = options.jitter || 0.2; + let label = options.label || 'operation'; + let shouldRetry = options.shouldRetry || isRetryable; + + let attempt = 0; + let delay = baseDelay; + + while(true){ + try{ + return await fn(); + }catch(error){ + attempt++; + if(!shouldRetry(error)){ + console.error(`${label} attempt ${attempt} failed permanently:`, error.message); + throw error; + } + + console.error(`${label} attempt ${attempt} failed, retrying in ${delay}ms:`, error.message); + let jittered = Math.round(delay * (1 + (Math.random() * 2 - 1) * jitter)); + await sleep(jittered); + delay = Math.min(delay * multiplier, maxDelay); + } + } +} + +// Convenience: retry a fetch call. On success returns the Response. +async function fetchWithRetry(url, options = {}, retryOptions = {}){ + return await withRetry(async () => { + let res = await fetch(url, options); + if(!res.ok){ + let error = new Error(`UpstreamError`); + error.message = `Request to ${url.split('?')[0]} failed( ${res.status})`; + error.status = res.status; + throw error; + } + return res; + }, { label: `fetch ${url.split('?')[0]}`, ...retryOptions }); +} + +// Convenience: retry a fetch call and parse JSON response. +async function fetchJSONWithRetry(url, options = {}, retryOptions = {}){ + let res = await fetchWithRetry(url, options, retryOptions); + return await res.json(); +} + +module.exports = { + withRetry, + fetchWithRetry, + fetchJSONWithRetry, + isRetryable, +}; diff --git a/controller/search.js b/controller/search.js index 89e1e79..45e21d5 100644 --- a/controller/search.js +++ b/controller/search.js @@ -5,6 +5,7 @@ const tmdb = require('>/controller/tmdb'); const { tmdbSearch, tmdbDetails } = tmdb; const library = require('>/controller/library'); const ollama = require('>/controller/ollama'); +const { fetchWithRetry, withRetry } = require('>/controller/retry'); // TPB numeric categories that count as Movies / TV( used for the q.php search and to // keep the LLM from ever seeing non-video results). @@ -79,13 +80,7 @@ function parseTPBHtml(html){ async function tpbSearch(title){ // Category 200 = all Video; we narrow to movie/tv subcats in prefilter(). let url = `${conf.search.tpbBase}/search/${encodeURIComponent(title)}/1/99/200`; - let res = await fetch(url, { headers: { 'User-Agent': 'Mozilla/5.0' } }); - if(!res.ok){ - let error = new Error('UpstreamError'); - error.message = `TPB search failed( ${res.status})`; - error.status = 502; - throw error; - } + let res = await fetchWithRetry(url, { headers: { 'User-Agent': 'Mozilla/5.0' } }, { label: 'tpb search' }); return parseTPBHtml(await res.text()); } @@ -168,9 +163,12 @@ async function rankReleases(candidates, meta){ imdb: t.imdb || null, })); - let parsed = await ollama.chatJSON( - buildSystemPrompt(meta.title, meta.year, meta.today), - JSON.stringify(compact) + let parsed = await withRetry( + () => ollama.chatJSON( + buildSystemPrompt(meta.title, meta.year, meta.today), + JSON.stringify(compact) + ), + { label: 'ollama rankReleases' } ); let options = Array.isArray(parsed.options) ? parsed.options : []; if(!options.length) throw new Error('Model returned no options'); @@ -278,7 +276,10 @@ function buildSeasonPrompt(title, season, today){ async function bestSeasonPack(candidates, meta){ let compact = candidates.map(t => ({ name: t.name, sizeHuman: humanSize(t.size), seeders: Number(t.seeders), info_hash: t.info_hash })); - let parsed = await ollama.chatJSON(buildSeasonPrompt(meta.title, meta.season, meta.today), JSON.stringify(compact)); + let parsed = await withRetry( + () => ollama.chatJSON(buildSeasonPrompt(meta.title, meta.season, meta.today), JSON.stringify(compact)), + { label: 'ollama bestSeasonPack' } + ); return parsed && parsed.info_hash ? parsed : null; } @@ -315,7 +316,10 @@ async function bestEpisodeTorrent(candidates, meta){ `Prefer HEVC/x265, 1080p, English audio, reasonable size. Reject CAM/TS/telesync and non-English-only.`, `Return STRICT JSON {"info_hash":"","why":"one short sentence"}; if nothing suitable, {"info_hash":null}.`, ].join(' '); - let parsed = await ollama.chatJSON(prompt, JSON.stringify(compact)); + let parsed = await withRetry( + () => ollama.chatJSON(prompt, JSON.stringify(compact)), + { label: 'ollama bestEpisodeTorrent' } + ); return parsed && parsed.info_hash ? parsed : null; } diff --git a/controller/tmdb.js b/controller/tmdb.js index d55953d..84efc5d 100644 --- a/controller/tmdb.js +++ b/controller/tmdb.js @@ -1,23 +1,13 @@ 'use strict'; const conf = require('>/conf'); - -async function fetchJSON(url, options){ - let res = await fetch(url, options); - if(!res.ok){ - let error = new Error('UpstreamError'); - error.message = `Request to ${url.split('?')[0]} failed( ${res.status})`; - error.status = 502; - throw error; - } - return await res.json(); -} +const { fetchJSONWithRetry } = require('./retry'); async function tmdbSearch(query){ let url = `https://api.themoviedb.org/3/search/multi?include_adult=false` + `&query=${encodeURIComponent(query)}&api_key=${conf.tmdb.apiKey}`; - let data = await fetchJSON(url); + let data = await fetchJSONWithRetry(url); return (data.results || []) .filter(item => item.media_type === 'movie' || item.media_type === 'tv') @@ -38,7 +28,7 @@ async function tmdbDetails(tmdbId, mediaType){ let url = `https://api.themoviedb.org/3/${mediaType}/${tmdbId}` + `?append_to_response=external_ids&api_key=${conf.tmdb.apiKey}`; - let data = await fetchJSON(url); + let data = await fetchJSONWithRetry(url); let date = data.release_date || data.first_air_date || ''; return { @@ -71,7 +61,7 @@ async function tmdbFindBest(title, year, mediaType){ // Per-season episode counts for a show( season 0 / specials excluded). async function tmdbSeasons(tmdbId){ - let data = await fetchJSON(`https://api.themoviedb.org/3/tv/${tmdbId}?api_key=${conf.tmdb.apiKey}`); + let data = await fetchJSONWithRetry(`https://api.themoviedb.org/3/tv/${tmdbId}?api_key=${conf.tmdb.apiKey}`); let date = data.first_air_date || ''; return { title: data.name, @@ -82,4 +72,4 @@ async function tmdbSeasons(tmdbId){ }; } -module.exports = { fetchJSON, tmdbSearch, tmdbDetails, tmdbFindBest, tmdbSeasons }; +module.exports = { tmdbSearch, tmdbDetails, tmdbFindBest, tmdbSeasons }; diff --git a/models/sql/torrent.js b/models/sql/torrent.js index 0d0a5b3..a2e49c9 100644 --- a/models/sql/torrent.js +++ b/models/sql/torrent.js @@ -2,18 +2,33 @@ const Transmission = require('transmission-promise'); const conf = require('>/conf'); +const { withRetry } = require('>/controller/retry'); -const tr_client = new Transmission(conf.transmission) +const tr_client = new Transmission(conf.transmission); + +// Wrap the raw Transmission client so every RPC call retries on transient network errors. +const trClient = new Proxy(tr_client, { + get(target, prop){ + let value = target[prop]; + if(typeof value !== 'function') return value; + return async function(...args){ + return await withRetry( + () => value.apply(target, args), + { label: `transmission ${String(prop)}` } + ); + }; + } +}); const statusMap = [ 'STOPPED', // 0 - 'CHECK_WAIT', // 1 + 'CHECK_WAIT', // 1 'CHECK', // 2 - 'DOWNLOAD_WAIT', // 3 - 'DOWNLOAD', // 4 - 'SEED_WAIT', // 5 - 'SEED', // 6 - 'ISOLATED', // 7 + 'DOWNLOAD_WAIT', // 3 + 'DOWNLOAD', // 4 + 'SEED_WAIT', // 5 + 'SEED', // 6 + 'ISOLATED', // 7 ]; module.exports = (sequelize, DataTypes, Model) => { @@ -27,7 +42,7 @@ module.exports = (sequelize, DataTypes, Model) => { // define association here } - static trClient = tr_client; + static trClient = trClient; // Map a raw TPB category id( e.g. 207) to one of our buckets. TPB groups by the // hundreds digit; TV shows are the exception pulled out of the Video group. @@ -57,7 +72,7 @@ module.exports = (sequelize, DataTypes, Model) => { 'download-dir': data.isPrivate ? `${conf.privateDownloadLocation}/${data.added_by}` : undefined, }; - let res = await tr_client.addUrl(data.magnetLink, options); + let res = await trClient.addUrl(data.magnetLink, options); return await super.create({ magnetLink: data.magnetLink, @@ -80,7 +95,7 @@ module.exports = (sequelize, DataTypes, Model) => { if(this.percentDone === 1) return this.dataValues - let res = ( await tr_client.get(this.hashString, [ + let res = ( await trClient.get(this.hashString, [ "eta", "percentDone", "status", "rateDownload", "errorString", "hashString", 'name', 'downloadDir', @@ -104,7 +119,7 @@ module.exports = (sequelize, DataTypes, Model) => { throw e } // console.error(`Torrent ${this.hashString} getTorrentData error`, error); - throw error; + throw error } }