Throttle stale pending block retries

MoneroOcean · MoneroOcean · commit a61883c721e9 · 2026-05-23T19:34:21.000Z
diff --git a/deployment/base.sql b/deployment/base.sql
@@ -390,6 +390,8 @@ INSERT INTO pool.config (module, item, item_value, item_type, Item_desc) VALUES
 INSERT INTO pool.config (module, item, item_value, item_type, Item_desc) VALUES ('email', 'workerRestartBody', 'Restarted worker module!', 'string', 'Body of worker restart admin email');
 INSERT INTO pool.config (module, item, item_value, item_type, Item_desc) VALUES ('email', 'remoteShareLmdbSubject', 'remote_share rejecting new work due to LMDB full', 'string', 'Subject of remote_share LMDB full admin email');
 INSERT INTO pool.config (module, item, item_value, item_type, Item_desc) VALUES ('email', 'remoteShareLmdbBody', 'remote_share is rejecting new share and block frames after LMDB reported map full while %(scope)s: %(detail)s.', 'string', 'Body of remote_share LMDB full admin email');
+INSERT INTO pool.config (module, item, item_value, item_type, Item_desc) VALUES ('email', 'remoteShareStalePendingSubject', 'FYI: Pending blocks not verified for over a month', 'string', 'Subject of remote_share stale pending block admin email');
+INSERT INTO pool.config (module, item, item_value, item_type, Item_desc) VALUES ('email', 'remoteShareStalePendingBody', 'remote_share has %(count)s pending block(s) older than %(age_days)s days.\n\n%(jobs)s\n\nPlease verify wallet/daemon sync and pending_blocks.', 'string', 'Body of remote_share stale pending block admin email');
 INSERT INTO pool.config (module, item, item_value, item_type, Item_desc) VALUES ('email', 'blockMgrBalanceSubject', 'block_manager unable to make balance increase', 'string', 'Subject of block_manager balance admin email');
 INSERT INTO pool.config (module, item, item_value, item_type, Item_desc) VALUES ('email', 'blockMgrBalanceBody', 'The block_manager module has hit an issue making a balance increase: %(message)s.  Please investigate and restart block_manager as appropriate', 'string', 'Body of block_manager balance admin email');
 INSERT INTO pool.config (module, item, item_value, item_type, Item_desc) VALUES ('email', 'blockMgrPaymentSubject', 'block_manager unable to make blockPayments', 'string', 'Subject of block_manager payment admin email');
diff --git a/lib/common/support.js b/lib/common/support.js
@@ -117,6 +117,8 @@ const DEFAULT_EMAIL_TEMPLATES = {
     workerRestartBody: "Restarted worker module!",
     remoteShareLmdbSubject: "remote_share rejecting new work due to LMDB full",
     remoteShareLmdbBody: "remote_share is rejecting new share and block frames after LMDB reported map full while %(scope)s: %(detail)s.",
+    remoteShareStalePendingSubject: "FYI: Pending blocks not verified for over a month",
+    remoteShareStalePendingBody: "remote_share has %(count)s pending block(s) older than %(age_days)s days.\n\n%(jobs)s\n\nPlease verify wallet/daemon sync and pending_blocks.",
     blockMgrBalanceSubject: "block_manager unable to make balance increase",
     blockMgrBalanceBody: "The block_manager module has hit an issue making a balance increase: %(message)s.  Please investigate and restart block_manager as appropriate",
     blockMgrPaymentSubject: "block_manager unable to make blockPayments",
diff --git a/lib/remote_share/pending_jobs.js b/lib/remote_share/pending_jobs.js
@@ -9,6 +9,13 @@ const ORPHAN_TRACK_TTL_MS = 12 * 60 * 60 * 1000;
 const LOG_STATE_TTL_MS = 12 * 60 * 60 * 1000;
 const DEFAULT_RETRY_DELAY_MS = 30 * 1000;
 const DEFAULT_MAX_RETRY_DELAY_MS = 30 * 60 * 1000;
+const DAY_MS = 24 * 60 * 60 * 1000;
+const DEFAULT_STALE_PENDING_BLOCK_AGE_MS = 30 * DAY_MS;
+const DEFAULT_STALE_PENDING_BLOCK_ALERT_CHECK_MS = DAY_MS;
+const DEFAULT_STALE_PENDING_BLOCK_ALERT_COOLDOWN_MS = DAY_MS;
+const DEFAULT_STALE_PENDING_BLOCK_SAMPLE_LIMIT = 20;
+const DEFAULT_STALE_PENDING_BLOCK_PROCESS_LIMIT = 5;
+const DEFAULT_STALE_PENDING_BLOCK_RETRY_DELAY_MS = 6 * 60 * 60 * 1000;
 
 function isOrphanHeader(header) {
     if (!header) return false;
@@ -56,6 +63,52 @@ function withDetail(fields, err, header) {
     return fields;
 }
 
+function renderEmailTemplate(item, values, fallback) {
+    if (global.support && typeof global.support.renderEmailTemplate === "function") return global.support.renderEmailTemplate(item, values, fallback);
+    const template = global.config && global.config.email && typeof global.config.email[item] === "string" ? global.config.email[item] : fallback;
+    return global.support && typeof global.support.formatTemplate === "function"
+        ? global.support.formatTemplate(template || "", values || {})
+        : String(template || "").replace(/%\(([^)]+)\)s/g, function replaceValue(_match, key) {
+            return values && Object.prototype.hasOwnProperty.call(values, key) ? String(values[key]) : "";
+        });
+}
+
+function safeDecodePendingPayload(job) {
+    if (!job || typeof job.payload !== "string" || !global.protos) return null;
+    try {
+        if (job.type === "block") return global.protos.Block.decode(Buffer.from(job.payload, "base64"));
+        if (job.type === "altblock") return global.protos.AltBlock.decode(Buffer.from(job.payload, "base64"));
+    } catch (_error) {}
+    return null;
+}
+
+function parseAltPortFromJobKey(key) {
+    if (typeof key !== "string") return null;
+    const parts = key.split(":");
+    if (parts.length < 2) return null;
+    const port = parseInt(parts[1], 10);
+    return Number.isFinite(port) ? port : null;
+}
+
+function formatStalePendingBlock(job, timeNow) {
+    const decoded = safeDecodePendingPayload(job);
+    const fields = ["type=" + job.type];
+    if (job.type === "altblock") {
+        const port = decoded && decoded.port ? decoded.port : parseAltPortFromJobKey(job.key);
+        if (port) fields.push("chain=" + formatCoinPort(port));
+        if (decoded && decoded.height) fields.push("height=" + decoded.height);
+    } else {
+        fields.push("chain=" + formatCoinPort(global.config && global.config.daemon ? global.config.daemon.port : 0));
+        if (typeof job.blockId !== "undefined") fields.push("blockId=" + job.blockId);
+    }
+    if (decoded && decoded.hash) fields.push("hash=" + decoded.hash);
+    else if (typeof job.key === "string") fields.push("job=" + job.key);
+    fields.push("ageDays=" + Math.floor(Math.max(0, timeNow - job.createdAt) / DAY_MS));
+    fields.push("attempts=" + (job.attempts || 0));
+    if (job.lastError) fields.push("lastError=" + job.lastError);
+    return fields.join(" ");
+}
+
 function createLmdbStorage(database) {
     let pendingJobDb = null;
 
@@ -112,6 +165,7 @@ function createLmdbStorage(database) {
 
         loadDueJobs(timeNow, limit) {
             const jobsByKey = new Map();
+            let nextDueAt = null;
             const txn = database.env.beginTxn({ readOnly: true });
             try {
                 function collectJobs(key, value) {
@@ -120,6 +174,9 @@ function createLmdbStorage(database) {
                     try {
                         const job = JSON.parse(value);
                         if (job.nextAttemptAt <= timeNow) jobsByKey.set(key, job);
+                        else if (typeof job.nextAttemptAt === "number" && (nextDueAt === null || job.nextAttemptAt < nextDueAt)) {
+                            nextDueAt = job.nextAttemptAt;
+                        }
                     } catch (_error) {
                         jobsByKey.set(key, { key, type: "invalid", invalid: true });
                     }
@@ -134,6 +191,7 @@ function createLmdbStorage(database) {
             jobs.sort(function bySchedule(left, right) {
                 return left.nextAttemptAt - right.nextAttemptAt || left.createdAt - right.createdAt;
             });
+            jobs.nextDueAt = nextDueAt;
             return jobs;
         },
 
@@ -169,6 +227,12 @@ module.exports = function createPendingJobs(options) {
     const retryDelayMs = opts.retryDelayMs || DEFAULT_RETRY_DELAY_MS;
     const maxRetryDelayMs = opts.maxRetryDelayMs || DEFAULT_MAX_RETRY_DELAY_MS;
     const orphanGraceMs = opts.orphanGraceMs || 10 * 60 * 1000;
+    const stalePendingBlockAgeMs = typeof opts.stalePendingBlockAgeMs === "number" ? opts.stalePendingBlockAgeMs : DEFAULT_STALE_PENDING_BLOCK_AGE_MS;
+    const stalePendingBlockAlertCheckMs = typeof opts.stalePendingBlockAlertCheckMs === "number" ? opts.stalePendingBlockAlertCheckMs : DEFAULT_STALE_PENDING_BLOCK_ALERT_CHECK_MS;
+    const stalePendingBlockAlertCooldownMs = typeof opts.stalePendingBlockAlertCooldownMs === "number" ? opts.stalePendingBlockAlertCooldownMs : DEFAULT_STALE_PENDING_BLOCK_ALERT_COOLDOWN_MS;
+    const stalePendingBlockSampleLimit = typeof opts.stalePendingBlockSampleLimit === "number" ? opts.stalePendingBlockSampleLimit : DEFAULT_STALE_PENDING_BLOCK_SAMPLE_LIMIT;
+    const stalePendingBlockProcessLimit = typeof opts.stalePendingBlockProcessLimit === "number" ? opts.stalePendingBlockProcessLimit : DEFAULT_STALE_PENDING_BLOCK_PROCESS_LIMIT;
+    const stalePendingBlockRetryDelayMs = typeof opts.stalePendingBlockRetryDelayMs === "number" ? opts.stalePendingBlockRetryDelayMs : DEFAULT_STALE_PENDING_BLOCK_RETRY_DELAY_MS;
     const logger = opts.logger || console;
     const storage = opts.storage || createLmdbStorage(database);
 
@@ -182,7 +246,9 @@ module.exports = function createPendingJobs(options) {
         badAltBlocks: new Map(),
         potentiallyBadAltBlocks: new Map(),
         orphanSince: new Map(),
-        loggedStates: new Map()
+        loggedStates: new Map(),
+        lastStalePendingBlockAlertCheckAt: 0,
+        nextDueJobCheckAt: 0
     };
 
     function logState(jobKey, nextState, message) {
@@ -252,6 +318,7 @@ module.exports = function createPendingJobs(options) {
             const retryBackoffMs = retryDelayMs * retryMultiplier;
             nextRetryDelayMs = Math.min(retryBackoffMs, maxRetryDelayMs);
         }
+        if (retryOptions.stale === true) nextRetryDelayMs = Math.max(nextRetryDelayMs, stalePendingBlockRetryDelayMs);
         job.nextAttemptAt = Math.round(timeNow + nextRetryDelayMs);
         job.lastError = stateName;
         storage.save(job);
@@ -264,6 +331,87 @@ module.exports = function createPendingJobs(options) {
         clearState(job.key);
     }
 
+    function isStalePendingBlock(job, timeNow) {
+        if (!job || (job.type !== "block" && job.type !== "altblock")) return false;
+        if (typeof job.createdAt !== "number") return false;
+        return timeNow - job.createdAt >= stalePendingBlockAgeMs;
+    }
+
+    function saveStaleThrottle(job, timeNow) {
+        job.nextAttemptAt = Math.round(timeNow + stalePendingBlockRetryDelayMs);
+        job.lastError = "stale_pending_throttled";
+        storage.save(job);
+        logState(job.key, "stale_pending_throttled", formatLogEvent("Pending job", {
+            job: job.key,
+            status: "stale-throttled",
+            ageDays: Math.floor(Math.max(0, timeNow - job.createdAt) / DAY_MS)
+        }));
+    }
+
+    function selectJobsForProcessing(jobs, timeNow) {
+        const selected = [];
+        const staleLimit = Math.max(0, stalePendingBlockProcessLimit);
+        let staleSelected = 0;
+        for (const job of jobs) {
+            if (isStalePendingBlock(job, timeNow)) {
+                if (staleSelected >= staleLimit) {
+                    saveStaleThrottle(job, timeNow);
+                    continue;
+                }
+                staleSelected += 1;
+            }
+            selected.push(job);
+        }
+        return selected;
+    }
+
+    function maybeSendStalePendingBlockAlert(timeNow) {
+        if (state.lastStalePendingBlockAlertCheckAt && timeNow - state.lastStalePendingBlockAlertCheckAt < stalePendingBlockAlertCheckMs) return;
+        state.lastStalePendingBlockAlertCheckAt = timeNow;
+        if (!global.support || typeof global.support.sendAdminFyi !== "function") return;
+        if (!global.config || !global.config.general || !global.config.general.adminEmail) return;
+
+        let staleJobs;
+        try {
+            staleJobs = storage.loadAllJobs().filter(function findStalePendingBlock(job) {
+                return isStalePendingBlock(job, timeNow);
+            });
+        } catch (error) {
+            if (logger && typeof logger.log === "function") {
+                logger.log(formatLogEvent("Pending block alert", {
+                    status: "failed",
+                    detail: formatHeaderErrorDetail(error)
+                }));
+            }
+            return;
+        }
+        if (staleJobs.length === 0) return;
+
+        staleJobs.sort(function byAge(left, right) {
+            return left.createdAt - right.createdAt || String(left.key).localeCompare(String(right.key));
+        });
+
+        const sampleJobs = staleJobs.slice(0, Math.max(1, stalePendingBlockSampleLimit));
+        const jobLines = sampleJobs.map(function formatJob(job) {
+            return "- " + formatStalePendingBlock(job, timeNow);
+        });
+        if (staleJobs.length > sampleJobs.length) {
+            jobLines.push("- ... " + (staleJobs.length - sampleJobs.length) + " more pending block(s) omitted");
+        }
+
+        const values = {
+            count: staleJobs.length,
+            age_days: Math.floor(stalePendingBlockAgeMs / DAY_MS),
+            jobs: jobLines.join("\n")
+        };
+        global.support.sendAdminFyi(
+            "remote_share:stale-pending-blocks",
+            renderEmailTemplate("remoteShareStalePendingSubject", values, "FYI: Pending blocks not verified for over a month"),
+            renderEmailTemplate("remoteShareStalePendingBody", values, "remote_share has %(count)s pending block(s) older than %(age_days)s days.\n\n%(jobs)s\n\nPlease verify wallet/daemon sync and pending_blocks."),
+            { cooldownMs: stalePendingBlockAlertCooldownMs }
+        );
+    }
+
     function saveResolvedBlock(blockId, blockDataDecoded) {
         const shares = database.getCache(poolTypeStr(blockDataDecoded.poolType) + "_stats2");
         blockDataDecoded.shares = shares ? shares.roundHashes : 0;
@@ -355,9 +503,11 @@ module.exports = function createPendingJobs(options) {
             }
 
             if (err || !header || !header.reward) {
+                const retryOptions = { backoff: true };
+                if (isStalePendingBlock(job, timeNow)) retryOptions.stale = true;
                 saveRetry(job, "waiting_block_header", formatLogEvent("Pending block", withDetail(Object.assign({}, blockFields, {
                     status: "waiting-header-reward"
-                }), err, header)), { backoff: true });
+                }), err, header)), retryOptions);
                 return callback();
             }
 
@@ -419,7 +569,8 @@ module.exports = function createPendingJobs(options) {
 
             const profile = global.coinFuncs.getPoolProfile(blockDataDecoded.port);
             if (isWaitingForAltDepth(profile, header)) {
-                saveRetry(job, "waiting_for_depth", formatLogEvent("Pending altblock", Object.assign({}, altBlockFields, { status: "waiting-maturity" })));
+                const retryOptions = isStalePendingBlock(job, timeNow) ? { stale: true } : undefined;
+                saveRetry(job, "waiting_for_depth", formatLogEvent("Pending altblock", Object.assign({}, altBlockFields, { status: "waiting-maturity" })), retryOptions);
                 return callback();
             }
 
@@ -431,9 +582,11 @@ module.exports = function createPendingJobs(options) {
                 const badBlockMap = state.potentiallyBadAltBlocks.get(blockDataDecoded.port) || new Map();
                 badBlockMap.set(blockDataDecoded.hash, timeNow);
                 state.potentiallyBadAltBlocks.set(blockDataDecoded.port, badBlockMap);
+                const retryOptions = { backoff: true };
+                if (isStalePendingBlock(job, timeNow)) retryOptions.stale = true;
                 saveRetry(job, "waiting_altblock_header", formatLogEvent("Pending altblock", withDetail(Object.assign({}, altBlockFields, {
                     status: "waiting-header-reward"
-                }), err, header)), { backoff: true });
+                }), err, header)), retryOptions);
                 return callback();
             }
 
@@ -462,6 +615,7 @@ module.exports = function createPendingJobs(options) {
 
     return {
         enqueueBlock(blockId, payload, block) {
+            state.nextDueJobCheckAt = 0;
             storage.save({
                 key: "block:" + blockId + ":" + block.hash,
                 type: "block",
@@ -475,6 +629,7 @@ module.exports = function createPendingJobs(options) {
         },
 
         enqueueAltBlock(blockId, payload, block) {
+            state.nextDueJobCheckAt = 0;
             storage.save({
                 key: "alt:" + block.port + ":" + block.height + ":" + block.hash,
                 type: "altblock",
@@ -489,19 +644,28 @@ module.exports = function createPendingJobs(options) {
 
         processDueJobs() {
             if (state.processing || state.closed) return;
+            const timeNow = Date.now();
+            maybeSendStalePendingBlockAlert(timeNow);
+            if (state.nextDueJobCheckAt && timeNow < state.nextDueJobCheckAt) return;
             state.processing = true;
             try {
                 pruneState();
 
-                const jobs = storage.loadDueJobs(Date.now(), 100);
+                const jobs = storage.loadDueJobs(timeNow, 100);
+                if (jobs.length === 0 && typeof jobs.nextDueAt === "number") {
+                    state.nextDueJobCheckAt = jobs.nextDueAt;
+                } else {
+                    state.nextDueJobCheckAt = 0;
+                }
+                const jobsToProcess = selectJobsForProcessing(jobs, timeNow);
                 let index = 0;
                 const next = () => {
-                    if (index >= jobs.length) {
+                    if (index >= jobsToProcess.length) {
                         state.processing = false;
                         finishCloseIfIdle();
                         return;
                     }
-                    processJob(jobs[index++], next);
+                    processJob(jobsToProcess[index++], next);
                 };
                 next();
             } catch (error) {
diff --git a/tests/remote_share/block_jobs.js b/tests/remote_share/block_jobs.js
