Add exponential backoff retry to all downstream calls

New controller/retry.js wraps async calls with infinite retries, capped

at 30s with jitter. Applied to: TPB search, TMDB, Emby, Ollama chat,

Transmission RPC, and the status/status interval calls inherit it via

a Proxy around the Transmission client.

Also handles ollama package returning pre-parsed JSON objects.

Signed-off-by: William Mantly <wmantly@gmail.com>

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;
 }
 

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

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 };

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`

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,
+};

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":"<from candidates>","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;
 }
 

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 };

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
       }
     }