dedupe: add --dedupeConcurrent to be able to dedupe concurrent crawls (#1065)

- crawl cancelation with dedupeConcurrent: track of canceled concurrent
- crawls in 'canceledcrawls' key, to be treated as removed crawls and purged
- docs: update docs for --dedupeConcurrent trade-offs and update cli page
- dedupe: account for page urls that redirect by checking for finalPageUrl
- fixes #1064

---------

Co-authored-by: Tessa Walsh <tessa@bitarchivist.net>

Author: ikreymer

docs/docs/develop/dedupe.md

@@ -60,11 +60,20 @@ Once a crawl is finished, the merged index keys are updated as follows:
 
 - `allhashes`: `{[hash]: [crawlid]}` - The main merged index. For each hash in `h:${crawlid}`, an entry is added mapping `hash` to the `crawlid`, indicating the hash is found in this crawl. The full data from `h:${crawlid}` is not copied to save space.
 
-- `allcounts`: A sum of all the `h:${crawlid}:counts` for all crawlids in `allcrawls`. The `allcounts` fields are incremented by each new `h:${crawlid}:counts`
+- `allcounts`: A sum of all the `h:${crawlid}:counts` for all crawl ids in `allcrawls`. The `allcounts` fields are incremented by each new `h:${crawlid}:counts`.
+
+- `canceledcrawls`: Use with concurrent crawls only to store crawl ids of crawls that were cancelled or have failed
+while being deduped. If this key exists, indicates there may be data missing, if it has not been re-crawled.
 
 Committing the crawl may take a long time, depending on the size of the crawl. For this reason, the indexer entrypoint
 includes a dedicated command to commit a single crawl, `indexer --commitCrawlId <crawlid>`.
 
+### Concurrent Crawl Support
+
+If running with `--dedupeConcurrent`, each new hash is automatically committed to `allhashes` and `allcounts` is updated
+for every new hash. If a concurrent crawl is canceled, it is additionally tracked in `canceledcrawls` to indicate that it
+needs to be included in the removed crawl count.
+
 #### Dedupe Lookup
 
 To write a revisit record, two Redis lookups are needed:

docs/docs/user-guide/cli-options.md

@@ -203,6 +203,13 @@ Options:
                                             duplicate pages can be skipped. -1 m
                                             eans never skip duplicate pages
                                                           [number] [default: -1]
+      --dedupeConcurrent                    If set, immediately add deduped URLs
+                                             to the main dedupe index instead of
+                                             to uncommittedcrawls. This allows d
+                                            edupe of concurrently running crawls
+                                            , but makes cancelling or failed cra
+                                            wls potentially lossy
+                                                      [boolean] [default: false]
       --saveState                           If the crawl state should be seriali
                                             zed to the crawls/ directory. Defaul
                                             ts to 'partial', only saved when cra

docs/docs/user-guide/dedupe.md

@@ -26,11 +26,18 @@ been archived, it will not be saved again, instead a [`revisit` record will be c
 
     While the index needs to be Redis-compatible, any Redis-compatible database can be used as well without additional changes. [Apache Kvrocks](https://kvrocks.apache.org/) is a good choice for the dedupe index database as it persists the data on disk, instead of keeping it all in memory like Redis. 
 
-When multiple crawls are running at the same time, the resources from one crawl are not yet available to the other crawls.
-This is to account for crawls that may be cancelled or fail.
+
+### Crawl Cancellation and Concurrent Crawl support
+
+By default, when multiple crawls are running at the same time, the resources from one crawl are not yet available to the other crawls.
+This is to account for crawls that may be cancelled or fail. If the crawl is cancelled/failed, the data is not entered into the shared dedupe index.
 
 Once a crawl is complete, its data is fully 'committed' to the index and available to be deduplicated against by future crawls. This happens automatically when running the crawler via command line but can also be triggered [via a special indexer command](#committing-finished-crawls-to-the-index).
 
+Sometimes it is necessary to deduplicate multiple concurrently running crawls. The `--dedupeConcurrent` flag can be used to ensure
+the dedupe index is updated immediately while the crawl is running. As a trade-off, there may be data loss if the crawl is cancelled or fails, as other crawls may deduplicate against records from a crawl whose data is discarded and not stored.
+For this reason, concurrent deduplication is not the default behavior, but may be enabled if concurrent crawls will not be canceled and will be retried on failure.
+
 ### Example: Running crawls with deduplication
 
 Here's a quick example of running crawls with dedupe against the same index:
@@ -98,9 +105,13 @@ docker run -it webrecorder/browsertrix-crawler indexer --commitCrawlId <crawl-id
 
 The commit process may take a long time if it was a large crawl, but generally should finish quickly.
 
+If running with `--dedupeConcurrent`, no commit operation is needed as the data is automatically committed.
+
 ### Handling interrupted or canceled crawls
 
-If a crawler is interrupted, eg. with SIGINT, the dedupe data stored for that crawl will not yet committed.
+If a crawler is interrupted, eg. with SIGINT, the dedupe data stored for that crawl will not yet be committed
+(unless `--dedupeConcurrent` was used, see below).
+
 Since the crawl has not finished, the user may or may not want to include the data in the dedupe index.
 
 To include the partially finished crawl, run the above command with `--commitCrawlId`.
@@ -114,6 +125,14 @@ docker run -it webrecorder/browsertrix-crawler indexer --cancelCrawlId <crawl-id
 This operational is not required - the data from the interrupted crawl will not be used in further dedupe, it will
 simply free up the data on the Redis.
 
+#### With concurrent crawl support
+
+If the crawl was run with `--dedupeConcurrent` and is cancelled or fails, the crawl is tracked in a list of removed crawls that
+should be purged. The crawl's data will be removed from the index when the `indexer --remove` operation is performed.
+
+Since other crawl data may have been deduplicated against the failed crawl, it is possible that cancelled or failed crawls may result in missing data
+if the crawl is marked as dependency of another crawl. Rerunning the crawl may patch the missing data in this case.
+
 
 ## Page Deduplication
 

src/crawler.ts

@@ -399,6 +399,7 @@ export class Crawler {
       os.hostname(),
       this.params.maxPageRetries,
       dedupeRedis,
+      this.params.dedupeConcurrent,
       this.params.rateLimitTimeout,
       this.params.rateLimitInterruptCount,
       this.params.rateLimitMaxRetries,
@@ -1875,7 +1876,7 @@ self.__bx_behaviors.selectMainBehavior();
       }
     }
     if (this.isExternalDedupeStore) {
-      await this.crawlState.addUncommited();
+      await this.crawlState.addCrawlForDedupe();
     }
 
     if (POST_CRAWL_STATES.includes(initState)) {

src/indexer.ts

@@ -87,7 +87,7 @@ export class CrawlIndexer {
     const params = this.initArgs();
 
     const redis = await initRedisWaitForSuccess(params.redisDedupeUrl);
-    const dedupeIndex = new RedisDedupeIndex(redis, "");
+    const dedupeIndex = new RedisDedupeIndex(redis, "", false);
 
     if (params.commitCrawlId) {
       // Commit one crawl and exit
@@ -97,11 +97,20 @@ export class CrawlIndexer {
       await dedupeIndex.commitDedupeDone(params.commitCrawlId);
       process.exit(ExitCodes.Success);
     } else if (params.cancelCrawlId) {
-      // Cancel one crawl and exit
-      logger.info("Deleting data for cancelled uncommitted crawl", {
-        crawlId: params.cancelCrawlId,
-      });
-      await dedupeIndex.clearUncommitted(params.cancelCrawlId);
+      if (await dedupeIndex.clearUncommitted(params.cancelCrawlId)) {
+        // Cancel one crawl and exit
+        logger.info("Deleting data for cancelled uncommitted crawl", {
+          crawlId: params.cancelCrawlId,
+        });
+      } else if (await dedupeIndex.markCanceledCrawl(params.cancelCrawlId)) {
+        logger.info("Crawl already committed, mark crawl for removal", {
+          crawlId: params.cancelCrawlId,
+        });
+      } else {
+        logger.info("Canceled crawl id not found, exiting", {
+          crawlId: params.cancelCrawlId,
+        });
+      }
       process.exit(ExitCodes.Success);
     } else if (!params.sourceUrl) {
       await logger.fatal(

src/util/argParser.ts

@@ -476,6 +476,13 @@ class ArgParser {
           default: -1,
         },
 
+        dedupeConcurrent: {
+          describe:
+            "If set, immediately add deduped URLs to the main dedupe index instead of to uncommittedcrawls. This allows dedupe of concurrently running crawls, but makes cancelling or failed crawls potentially lossy",
+          type: "boolean",
+          default: false,
+        },
+
         saveState: {
           describe:
             "If the crawl state should be serialized to the crawls/ directory. Defaults to 'partial', only saved when crawl is interrupted",

src/util/constants.ts

@@ -26,6 +26,7 @@ export const DUPE_ALL_HASH_KEY = "alldupes";
 export const DUPE_ALL_CRAWLS = "allcrawls";
 export const DUPE_ALL_COUNTS = "allcounts";
 export const DUPE_UNCOMMITTED = "uncommittedcrawls";
+export const DUPE_CANCELED_CRAWLS = "canceledcrawls";
 
 export enum BxFunctionBindings {
   BehaviorLogFunc = "__bx_log",

src/util/recorder.ts

@@ -893,7 +893,7 @@ export class Recorder extends EventEmitter {
     // write a revisit record, track pages as a duplicate, and abort the page
     if (
       !streamingConsume &&
-      url === this.pageUrl &&
+      url === this.finalPageUrl &&
       reqresp.payload &&
       this.dedupePagesMinDepth >= 0 &&
       this.pageSeedDepth >= this.dedupePagesMinDepth

src/util/state.ts

@@ -10,6 +10,7 @@ import {
   DUPE_ALL_HASH_KEY,
   DUPE_ALL_CRAWLS,
   DUPE_ALL_COUNTS,
+  DUPE_CANCELED_CRAWLS,
   DUPE_UNCOMMITTED,
   CrawlStatus,
   SkippedReason,
@@ -250,9 +251,12 @@ export class RedisDedupeIndex {
 
   noremove = "noremove:" + Date.now();
 
-  constructor(dedupeRedis: Redis, crawlId: string) {
+  autoCommit = false;
+
+  constructor(dedupeRedis: Redis, crawlId: string, autoCommit: boolean) {
     this.dedupeRedis = dedupeRedis;
     this.crawlId = crawlId;
+    this.autoCommit = autoCommit;
   }
 
   // DEDUPE SOURCE WACZ (to track dependencies)
@@ -384,20 +388,34 @@ export class RedisDedupeIndex {
     logger.debug("Crawl counts added to allcounts", {}, "dedupe");
   }
 
-  // ADD UNCOMITTED CRAWL
-  async addUncommited() {
+  // ADD UNCOMMITTED CRAWL
+  async addCrawlForDedupe() {
     if (this.crawlId) {
-      await this.dedupeRedis.sadd(DUPE_UNCOMMITTED, this.crawlId);
+      // if autoCommit, add directly to all crawls list
+      if (this.autoCommit) {
+        await this.dedupeRedis.sadd(DUPE_ALL_CRAWLS, this.crawlId);
+      } else {
+        await this.dedupeRedis.sadd(DUPE_UNCOMMITTED, this.crawlId);
+      }
     }
   }
 
   // CLEAR UNCOMMITTED
-  async clearUncommitted(crawlId?: string) {
+  async clearUncommitted(crawlId?: string): Promise<boolean> {
+    // won't be in the uncommitted list as auto-committing
+    if (this.autoCommit) {
+      return false;
+    }
+
     crawlId ||= this.crawlId;
     if (crawlId) {
-      await this.dedupeRedis.srem(DUPE_UNCOMMITTED, crawlId);
-      await this.deleteCrawlDedupeKeys(crawlId);
+      if ((await this.dedupeRedis.srem(DUPE_UNCOMMITTED, crawlId)) > 0) {
+        await this.deleteCrawlDedupeKeys(crawlId);
+        return true;
+      }
     }
+
+    return false;
   }
 
   async clearAllUncommitted() {
@@ -458,11 +476,21 @@ export class RedisDedupeIndex {
     if (!origRecSize) {
       const { key, val } = this.getHashValue(hash, url, date, size);
       pipe.hsetnx(rootKey, key, val);
+
+      if (this.autoCommit) {
+        pipe.hsetnx(DUPE_ALL_HASH_KEY, key, this.crawlId);
+      }
     }
     this.incrTotalUrls(pipe, statsKey);
+    if (this.autoCommit) {
+      this.incrTotalUrls(pipe, DUPE_ALL_COUNTS);
+    }
 
     if (origRecSize) {
       this.incrDeduped(pipe, statsKey, origRecSize - size);
+      if (this.autoCommit) {
+        this.incrDeduped(pipe, DUPE_ALL_COUNTS, origRecSize - size);
+      }
     }
 
     await pipe.exec();
@@ -690,21 +718,45 @@ export class RedisDedupeIndex {
 
   async purgeUnusedCrawls() {
     const noRemoveSet = new Set<string>(
-      await this.dedupeRedis.smembers(this.noremove),
+      await this.dedupeRedis.sdiff(this.noremove, DUPE_CANCELED_CRAWLS),
     );
 
     await this.clearAndReadd(noRemoveSet);
 
     await this.dedupeRedis.del(this.noremove);
+    await this.dedupeRedis.del(DUPE_CANCELED_CRAWLS);
+  }
+
+  async markCanceledCrawl(removeCrawlId: string) {
+    if (!(await this.dedupeRedis.sismember(DUPE_ALL_CRAWLS, removeCrawlId))) {
+      return false;
+    }
+
+    await this.dedupeRedis.sadd(DUPE_CANCELED_CRAWLS, removeCrawlId);
+
+    await this.countUnusedCrawls();
+
+    return true;
   }
 
   async countUnusedCrawls() {
-    const removable = await this.dedupeRedis.sdiff(
-      DUPE_ALL_CRAWLS,
-      this.noremove,
-    );
+    const removeListCanceled =
+      await this.dedupeRedis.smembers(DUPE_CANCELED_CRAWLS);
 
-    await this.dedupeRedis.del(this.noremove);
+    let removable: Set<string>;
+
+    if (await this.dedupeRedis.scard(this.noremove)) {
+      const removeList = await this.dedupeRedis.sdiff(
+        DUPE_ALL_CRAWLS,
+        this.noremove,
+      );
+
+      await this.dedupeRedis.del(this.noremove);
+
+      removable = new Set<string>([...removeList, ...removeListCanceled]);
+    } else {
+      removable = new Set<string>(removeListCanceled);
+    }
 
     let total = 0;
 
@@ -722,7 +774,7 @@ export class RedisDedupeIndex {
     await this.dedupeRedis.hset(
       DUPE_ALL_COUNTS,
       "removedCrawls",
-      removable.length,
+      removable.size,
     );
     await this.dedupeRedis.hset(DUPE_ALL_COUNTS, "removedCrawlSize", total);
   }
@@ -817,11 +869,12 @@ export class RedisCrawlState extends RedisDedupeIndex {
     uid: string,
     maxRetries?: number,
     dedupeRedis?: Redis,
+    dedupeAutoCommit = false,
     rateLimitTTL?: number,
     rateLimitInterruptCount?: number,
     rateLimitMaxRetries?: number,
   ) {
-    super(dedupeRedis || redis, key);
+    super(dedupeRedis || redis, key, dedupeAutoCommit);
     this.redis = redis;
 
     this.uid = uid;