Add Indexer options to commit/cancel single crawl (#978)

Add a --commitCrawlId and --cancelCrawlId to indexer which will either:
- merge the the data for a single crawl into the main index. This can
take a variable amount of time, depending on size of index.
- improved logging for commit, and ensure counts are never incremented
twice, even if long-running commit job is restarted.
- delete data for canceled crawl from uncommitted crawls list (should be
very fast).

Both of these tasks also done as part of the crawl itself, and is still
useful for single-crawler operation. However, with Browsertrix
controlled in k8s, these tasks are better handled as separate jobs,
especially since crawl may be committed without restarting the crawler
container (eg. for paused crawls).

---------

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

Author: ikreymer

docs/docs/develop/dedupe.md

@@ -49,6 +49,8 @@ to the merged index. The following key is used to track uncomitted crawls:
 
 If a crawl is canceled, these 3 keys should be removed and the crawl id should be removed from `uncommittedcrawls`
 
+The indexer entrypoint includes a dedicated `indexer --cancelCrawlId <crawlid>` option to perform the cancellation operation.
+
 ### Committed Crawls / Merged Index Keys
 
 Once a crawl is finished, the merged index keys are updated as follows:
@@ -60,6 +62,9 @@ Once a crawl is finished, the merged index keys are updated as follows:
 
 - `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`
 
+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>`.
+
 #### Dedupe Lookup
 
 To write a revisit record, two Redis lookups are needed:
@@ -126,6 +131,13 @@ the existing crawls, by rerunning the crawl 'commit' process for each existing c
 
 The result is that all data related to removed crawls is purged, and the `removeCrawlSize` and `removeCrawls` counts are reset to 0.
 
+### Committing and Canceling
+
+The indexer entrypoint also includes options to commit or cancel a crawl, as explained above. These options,
+`--commitCrawlId <crawlid>` and `--cancelCrawlId <crawlid>` are useful when the crawler is controlled via k8s, e.g. with `--restartsOnError` set.
+
+When the `--restartsOnError` flag is set, it is expected that the commit/cancel operations will be run separately, and so they are not run as part of the crawl. Otherwise, they are attempted as part of the crawl (after WACZ upload or upon cancellation).
+
 ## Crawl Dependency Tracking in WACZ datapackage.json
 
 Deduplication adds an inherent dependency system, where a WACZ from one crawl (which contains `revisit` records) is now dependent on a WACZ from another crawl (which contains the original `response` records).

src/crawler.ts

@@ -1681,7 +1681,7 @@ self.__bx_behaviors.selectMainBehavior();
   }
 
   async cleanupOnCancel() {
-    if (this.isExternalDedupeStore) {
+    if (this.isExternalDedupeStore && !this.params.restartsOnError) {
       await this.crawlState.clearUncommitted();
     }
   }
@@ -2028,9 +2028,9 @@ self.__bx_behaviors.selectMainBehavior();
       return;
     }
 
-    if (this.isExternalDedupeStore) {
+    if (this.isExternalDedupeStore && !this.params.restartsOnError) {
       // commit crawl data to main index
-      logger.info("Committing dedupe index");
+      logger.info("Committing dedupe index", {}, "dedupe");
       await this.crawlState.commitDedupeDone();
     }
 

src/indexer.ts

@@ -44,7 +44,7 @@ export class CrawlIndexer {
         sourceUrl: {
           describe: "Source WACZ or Multi WACZ or Multi WACZ JSON to index",
           type: "string",
-          required: true,
+          required: false,
         },
 
         sourceCrawlId: {
@@ -59,12 +59,24 @@ export class CrawlIndexer {
           required: false,
           default: false,
         },
+
+        commitCrawlId: {
+          describe:
+            "If provided, commit single uncommitted crawl to merged index and exit",
+          type: "string",
+        },
+
+        cancelCrawlId: {
+          describe: "If provided, delete data for uncommitted crawl and exit",
+          type: "string",
+        },
       })
       .parseSync();
   }
 
   async run() {
     logger.setDebugLogging(true);
+    logger.setDefaultLogContext("dedupe");
 
     process.on("SIGINT", () => this.handleInterrupt("SIGINT"));
 
@@ -77,6 +89,27 @@ export class CrawlIndexer {
     const redis = await initRedisWaitForSuccess(params.redisDedupeUrl);
     const dedupeIndex = new RedisDedupeIndex(redis, "");
 
+    if (params.commitCrawlId) {
+      // Commit one crawl and exit
+      logger.info("Committing crawl to merged index", {
+        crawlId: params.commitCrawlId,
+      });
+      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);
+      process.exit(ExitCodes.Success);
+    } else if (!params.sourceUrl) {
+      logger.fatal(
+        "One of --commitCrawlId, --cancelCrawlId or --sourceUrl for import is required",
+      );
+      return;
+    }
+
     for await (const entry of this.iterWACZ({
       url: params.sourceUrl,
       name: basename(params.sourceUrl),

src/util/logger.ts

@@ -58,6 +58,7 @@ export const LOG_CONTEXT_TYPES = [
   "proxy",
   "scope",
   "robots",
+  "dedupe",
 ] as const;
 
 export type LogContext = (typeof LOG_CONTEXT_TYPES)[number];
@@ -79,9 +80,14 @@ class Logger {
   logLevels: string[] = [];
   contexts: LogContext[] = [];
   excludeContexts: LogContext[] = [];
+  defaultLogContext: LogContext = "general";
   crawlState?: RedisCrawlState | null = null;
   fatalExitCode: ExitCodes = ExitCodes.Fatal;
 
+  setDefaultLogContext(value: LogContext) {
+    this.defaultLogContext = value;
+  }
+
   setDefaultFatalExitCode(exitCode: number) {
     this.fatalExitCode = exitCode;
   }
@@ -185,19 +191,35 @@ class Logger {
     }
   }
 
-  info(message: string, data: unknown = {}, context: LogContext = "general") {
+  info(
+    message: string,
+    data: unknown = {},
+    context: LogContext = this.defaultLogContext,
+  ) {
     this.logAsJSON(message, data, context, "info");
   }
 
-  error(message: string, data: unknown = {}, context: LogContext = "general") {
+  error(
+    message: string,
+    data: unknown = {},
+    context: LogContext = this.defaultLogContext,
+  ) {
     this.logAsJSON(message, data, context, "error");
   }
 
-  warn(message: string, data: unknown = {}, context: LogContext = "general") {
+  warn(
+    message: string,
+    data: unknown = {},
+    context: LogContext = this.defaultLogContext,
+  ) {
     this.logAsJSON(message, data, context, "warn");
   }
 
-  debug(message: string, data: unknown = {}, context: LogContext = "general") {
+  debug(
+    message: string,
+    data: unknown = {},
+    context: LogContext = this.defaultLogContext,
+  ) {
     if (this.debugLogging) {
       this.logAsJSON(message, data, context, "debug");
     }
@@ -206,7 +228,7 @@ class Logger {
   fatal(
     message: string,
     data = {},
-    context: LogContext = "general",
+    context: LogContext = this.defaultLogContext,
     exitCode = ExitCodes.Success,
   ) {
     exitCode = exitCode || this.fatalExitCode;

src/util/state.ts

@@ -273,19 +273,32 @@ export class RedisDedupeIndex {
 
   // COMMIT DEDUPE TO SHARED INDEX
 
-  async commitDedupeDone(crawlId?: string) {
+  async commitDedupeDone(crawlId?: string, uncommitted_key = DUPE_UNCOMMITTED) {
     crawlId = crawlId || this.crawlId;
 
-    // set this first, as crawl is considered 'committed' as soon
-    // as data is partially merged in
-    await this.dedupeRedis.srem(DUPE_UNCOMMITTED, crawlId);
-    await this.dedupeRedis.sadd(DUPE_ALL_CRAWLS, crawlId);
+    // check if already committed
+    if (!(await this.dedupeRedis.sismember(uncommitted_key, crawlId))) {
+      logger.warn(
+        "Crawl not found for committing, or already committed",
+        {
+          crawlId,
+        },
+        "dedupe",
+      );
+      return;
+    }
+
+    let totalHashes = 0;
+    let newHashes = 0;
 
     for await (const hashes of this.dedupeRedis.hscanStream(`h:${crawlId}`)) {
       let isValue = false;
       for (const hash of hashes) {
         if (!isValue) {
-          await this.dedupeRedis.hsetnx(DUPE_ALL_HASH_KEY, hash, crawlId);
+          if (await this.dedupeRedis.hsetnx(DUPE_ALL_HASH_KEY, hash, crawlId)) {
+            newHashes++;
+          }
+          totalHashes++;
         }
         isValue = !isValue;
       }
@@ -307,8 +320,36 @@ export class RedisDedupeIndex {
       }
     }
 
-    // add counts
-    await this.addCrawlCounts(crawlId);
+    logger.info(
+      "Crawl committed to merged index!",
+      {
+        crawlId,
+        totalHashes,
+        newHashes,
+        numWacz,
+      },
+      "dedupe",
+    );
+
+    // move to all crawls here
+    if (
+      !(await this.dedupeRedis.smove(uncommitted_key, DUPE_ALL_CRAWLS, crawlId))
+    ) {
+      // if already moved, return here to avoid duplicating counts
+      logger.warn(
+        "Crawl counts not updated, already added to allcounts",
+        {},
+        "dedupe",
+      );
+      return;
+    }
+
+    const counts = await this.dedupeRedis.hgetall(`h:${crawlId}:counts`);
+    for (const [key, value] of Object.entries(counts)) {
+      await this.dedupeRedis.hincrby(DUPE_ALL_COUNTS, key, Number(value));
+    }
+
+    logger.debug("Crawl counts added to allcounts", {}, "dedupe");
   }
 
   // ADD UNCOMITTED CRAWL
@@ -319,10 +360,11 @@ export class RedisDedupeIndex {
   }
 
   // CLEAR UNCOMMITTED
-  async clearUncommitted() {
-    if (this.crawlId) {
-      await this.dedupeRedis.srem(DUPE_UNCOMMITTED, this.crawlId);
-      await this.deleteCrawlDedupeKeys(this.crawlId);
+  async clearUncommitted(crawlId?: string) {
+    crawlId ||= this.crawlId;
+    if (crawlId) {
+      await this.dedupeRedis.srem(DUPE_UNCOMMITTED, crawlId);
+      await this.deleteCrawlDedupeKeys(crawlId);
     }
   }
 
@@ -413,13 +455,6 @@ export class RedisDedupeIndex {
     pipe.hincrby(key, "totalCrawlSize", value);
   }
 
-  async addCrawlCounts(crawlId: string) {
-    const counts = await this.dedupeRedis.hgetall(`h:${crawlId}:counts`);
-    for (const [key, value] of Object.entries(counts)) {
-      await this.dedupeRedis.hincrby(DUPE_ALL_COUNTS, key, Number(value));
-    }
-  }
-
   // IMPORT
 
   async queueImportSource(id: string, data: string) {
@@ -674,8 +709,7 @@ export class RedisDedupeIndex {
     // readd all crawls that should be kept
     for (const crawlId of readdCrawls) {
       await this.setUpdateProgress(0.5 + 0.5 * (count++ / numCrawls));
-      await this.commitDedupeDone(crawlId);
-      await this.dedupeRedis.srem(TO_REMOVE_CRAWLS, crawlId);
+      await this.commitDedupeDone(crawlId, TO_REMOVE_CRAWLS);
     }
 
     // clear data for remaining