Add total timeout + limit redis queue retries (#248)

* time limits: readd total timeount to runTask() in worker, just in case
refactor working runTask() to either return true/false if task was timed out
if timed out, recreate the page
redis: add limit to retried URLs, currently set to 1
* retry: remove URL if not retrying, log removal of URL from queue

Author: ikreymer

crawler.js

@@ -22,6 +22,7 @@ import { parseArgs } from "./util/argParser.js";
 import { initRedis } from "./util/redis.js";
 import { Logger, errJSON, setExternalLogStream, setDebugLogging } from "./util/logger.js";
 import { WorkerPool } from "./util/worker.js";
+import { sleep, timedRun } from "./util/timing.js";
 
 import { getBrowserExe, loadProfile, chromeArgs, getDefaultUA, evaluateWithCLI } from "./util/browser.js";
 
@@ -160,7 +161,7 @@ export class Crawler {
         } catch (e) {
           //this.logger.fatal("Unable to connect to state store Redis: " + redisUrl);
           this.logger.warn(`Waiting for redis at ${redisUrl}`, {}, "state");
-          await this.sleep(3);
+          await sleep(3);
         }
       }
 
@@ -381,7 +382,7 @@ export class Crawler {
 
     if (!this.isInScope(data, logDetails)) {
       this.logger.info("Page no longer in scope", data);
-      return;
+      return true;
     }
 
     try {
@@ -443,7 +444,7 @@ export class Crawler {
         } else {
           const behaviorTimeout = this.params.behaviorTimeout / 1000;
 
-          const res = await this.timedRun(
+          const res = await timedRun(
             this.runBehaviors(page, logDetails),
             behaviorTimeout,
             "Behaviors timed out",
@@ -466,7 +467,10 @@ export class Crawler {
     } catch (e) {
       this.logger.error("Page Errored", {...errJSON(e), ...logDetails}, "pageStatus");
       await this.markPageFailed(page);
+      return false;
     }
+
+    return true;
   }
 
   async runBehaviors(page, logDetails) {
@@ -605,7 +609,7 @@ export class Crawler {
     while (initState === "debug") {
       this.logger.info("Paused for debugging, will continue after manual resume");
 
-      await this.sleep(60);
+      await sleep(60);
 
       initState = await this.crawlState.getStatus();
     }
@@ -668,6 +672,7 @@ export class Crawler {
       crawlState: this.crawlState,
       screencaster: this.screencaster,
       healthChecker: this.healthChecker,
+      totalTimeout: (this.params.behaviorTimeout + this.params.timeout) / 1000 + 60,
       task: (opts) => this.crawlPage(opts)
     });
 
@@ -860,7 +865,7 @@ export class Crawler {
 
     let isHTMLPage = true;
 
-    const isHTMLResult = await this.timedRun(
+    const isHTMLResult = await timedRun(
       this.isHTML(url),
       FETCH_TIMEOUT_SECS,
       "HEAD request to determine if URL is HTML page timed out",
@@ -869,7 +874,7 @@ export class Crawler {
     if (isHTMLResult && (isHTMLResult.value == false)) {
       isHTMLPage = false;
       try {
-        const captureResult = await this.timedRun(
+        const captureResult = await timedRun(
           this.directFetchCapture(url),
           FETCH_TIMEOUT_SECS,
           "Direct fetch capture attempt timed out",
@@ -978,7 +983,7 @@ export class Crawler {
     }
     // in case page starts loading via fetch/xhr immediately after page load,
     // we want to ensure we don't exit too early
-    await this.sleep(0.5);
+    await sleep(0.5);
 
     try {
       await page.waitForNetworkIdle({timeout: this.params.netIdleWait * 1000});
@@ -1005,7 +1010,7 @@ export class Crawler {
       const frames = page.__filteredFrames;
 
       const linkResults = await Promise.allSettled(
-        frames.map(frame => this.timedRun(
+        frames.map(frame => timedRun(
           frame.evaluate(loadFunc, selector, extract),
           PAGE_OP_TIMEOUT_SECS,
           "Link extraction timed out",
@@ -1063,12 +1068,12 @@ export class Crawler {
     try {
       this.logger.debug("Check CF Blocking", logDetails);
 
-      while (await this.timedRun(
+      while (await timedRun(
         page.$("div.cf-browser-verification.cf-im-under-attack"),
         PAGE_OP_TIMEOUT_SECS
       )) {
         this.logger.debug("Cloudflare Check Detected, waiting for reload...", logDetails);
-        await this.sleep(5.5);
+        await sleep(5.5);
       }
     } catch (e) {
       //this.logger.warn("Check CF failed, ignoring");
@@ -1213,28 +1218,10 @@ export class Crawler {
         break;
       }
 
-      await this.sleep(1);
+      await sleep(1);
     }
   }
 
-  timedRun(promise, seconds, message="Promise timed out", logDetails={}, context="general") {
-    // return Promise return value or log error if timeout is reached first
-    const timeout = seconds * 1000;
-
-    const rejectPromiseOnTimeout = (timeout) => {
-      return new Promise((resolve, reject) => {
-        setTimeout(() => (reject("timeout reached")), timeout);
-      });
-    };
-
-    return Promise.race([promise, rejectPromiseOnTimeout(timeout)])
-      .catch(() => this.logger.error(message, {"seconds": seconds, ...logDetails}, context));
-  }
-
-  sleep(seconds) {
-    return new Promise(resolve => setTimeout(resolve, seconds * 1000));
-  }
-
   async parseSitemap(url, seedId) {
     const sitemapper = new Sitemapper({
       url,

util/state.js

@@ -194,6 +194,8 @@ export class RedisCrawlState extends BaseState
     super();
     this.redis = redis;
 
+    this.maxRetryPending = 1;
+
     this._lastSize = 0;
 
     this.uid = uid;
@@ -267,9 +269,16 @@ local res = redis.call('get', KEYS[3]);
 if not res then
   local json = redis.call('hget', KEYS[1], ARGV[1]);
   if json then
-    redis.call('lpush', KEYS[2], json);
+    local data = cjson.decode(json);
+    data['retry'] = (data['retry'] or 0) + 1;
     redis.call('hdel', KEYS[1], ARGV[1]);
-    return 1
+    if tonumber(data['retry']) <= tonumber(ARGV[2]) then
+      json = cjson.encode(data);
+      redis.call('lpush', KEYS[2], json);
+      return 1;
+    else
+      return 2;
+    end
   end
 end
 return 0;
@@ -459,8 +468,15 @@ return 0;
     const pendingUrls = await this.redis.hkeys(this.pkey);
 
     for (const url of pendingUrls) {
-      if (await this.redis.requeue(this.pkey, this.qkey, this.pkey + ":" + url, url)) {
+      const res = await this.redis.requeue(this.pkey, this.qkey, this.pkey + ":" + url, url, this.maxRetryPending);
+      switch (res) {
+      case 1:
         logger.info(`Requeued: ${url}`);
+        break;
+
+      case 2:
+        logger.info(`Not requeuing anymore: ${url}`);
+        break;
       }
     }
   }

util/timing.js

@@ -0,0 +1,23 @@
+import { Logger } from "./logger.js";
+
+const logger = new Logger();
+
+export function sleep(seconds) {
+  return new Promise(resolve => setTimeout(resolve, seconds * 1000));
+}
+
+export function timedRun(promise, seconds, message="Promise timed out", logDetails={}, context="general") {
+  // return Promise return value or log error if timeout is reached first
+  const timeout = seconds * 1000;
+
+  const rejectPromiseOnTimeout = (timeout) => {
+    return new Promise((resolve, reject) => {
+      setTimeout(() => (reject("timeout reached")), timeout);
+    });
+  };
+
+  return Promise.race([promise, rejectPromiseOnTimeout(timeout)])
+    .catch(() => logger.error(message, {"seconds": seconds, ...logDetails}, context));
+}
+
+

util/worker.js

@@ -3,6 +3,7 @@ import PQueue from "p-queue";
 import puppeteer from "puppeteer-core";
 
 import { Logger, errJSON } from "./logger.js";
+import { sleep, timedRun } from "./timing.js";
 
 const logger = new Logger();
 
@@ -15,11 +16,10 @@ const MAX_REUSE = 5;
 // ===========================================================================
 export class Worker
 {
-  constructor(id, browser, task, puppeteerOptions, screencaster, healthChecker) {
+  constructor(id, browser, task, screencaster, healthChecker) {
     this.id = id;
     this.browser = browser;
     this.task = task;
-    this.puppeteerOptions = puppeteerOptions;
     this.screencaster = screencaster;
     this.healthChecker = healthChecker;
 
@@ -29,19 +29,24 @@ export class Worker
     this.startPage = "about:blank?_browsertrix" + Math.random().toString(36).slice(2);
   }
 
-  async initPage() {
-    if (this.page && ++this.reuseCount <= MAX_REUSE) {
-      logger.debug("Reusing page", {reuseCount: this.reuseCount}, "worker");
-      return this.page;
-    } else if (this.page) {
+  async closePage() {
+    if (this.page) {
       try {
         await this.page.close();
       } catch (e) {
         // ignore
       }
       this.page = null;
     }
+  }
 
+  async initPage() {
+    if (this.page && ++this.reuseCount <= MAX_REUSE) {
+      logger.debug("Reusing page", {reuseCount: this.reuseCount}, "worker");
+      return this.page;
+    } else {
+      await this.closePage();
+    }
     //open page in a new tab
     this.pendingTargets = new Map();
 
@@ -71,7 +76,7 @@ export class Worker
         break;
       } catch (err) {
         logger.warn("Error getting new page in window context", {"workerid": this.id, ...errJSON(err)}, "worker");
-        await sleep(500);
+        await sleep(0.5);
         logger.warn("Retry getting new page");
 
         if (this.healthChecker) {
@@ -94,25 +99,10 @@ export class Worker
 
     logger.info("Starting page", {"workerid": this.id, "page": url}, "worker");
 
-    let result;
-    let errorState;
-
-    await this.task({
+    return await this.task({
       page: this.page,
       data: urlData,
     });
-
-    if (errorState) {
-      return {
-        type: "error",
-        error: errorState,
-      };
-    }
-
-    return {
-      data: result,
-      type: "success",
-    };
   }
 }
 
@@ -126,6 +116,7 @@ export class WorkerPool
     this.crawlState = options.crawlState;
     this.screencaster = options.screencaster;
     this.healthChecker = options.healthChecker;
+    this.totalTimeout = options.totalTimeout || 1e4;
 
     this.task = options.task;
 
@@ -156,7 +147,6 @@ export class WorkerPool
       id,
       this.browser,
       this.task,
-      this.puppeteerOptions,
       this.screencaster,
       this.healthChecker
     );
@@ -173,7 +163,7 @@ export class WorkerPool
 
     // wait half a second and try again
     logger.info("Waiting for available worker", {}, "worker");
-    await sleep(500);
+    await sleep(0.5);
 
     return await this.getAvailableWorker();
 
@@ -196,22 +186,32 @@ export class WorkerPool
       return;
     }
 
-    const result = await worker.runTask(job);
+    const result = await timedRun(
+      worker.runTask(job),
+      this.totalTimeout,
+      "Page Worker Timeout",
+      {"workerid": worker.id},
+      "worker"
+    );
+
+    if (!result) {
+      logger.debug("Resetting failed page", {}, "worker");
+
+      await worker.closePage();
 
-    if (result.type === "error") {
       if (job.callbacks) {
-        job.callbacks.reject(result.error);
+        job.callbacks.reject("timed out");
       }
       if (this.healthChecker) {
         this.healthChecker.incError();
       }
-    } else if (result.type === "success") {
+    } else {
       if (this.healthChecker) {
         this.healthChecker.resetErrors();
       }
 
       if (job.callbacks) {
-        job.callbacks.resolve(result.data);
+        job.callbacks.resolve(result);
       }
     }
 
@@ -236,7 +236,7 @@ export class WorkerPool
       }
 
       // wait half a second
-      await sleep(500);
+      await sleep(0.5);
     }
 
     logger.debug("Awaiting queue onIdle()", {}, "worker");
@@ -259,7 +259,4 @@ export class WorkerPool
   }
 }
 
-function sleep(millis) {
-  return new Promise(resolve => setTimeout(resolve, millis));
-}