Fixes off-by-1 error in WACZ compressed cdx index creation. (#1122)

Fixes off-by-1 error where the index.idx actually references the second
line in the index.cdx.gz block, and the last line was getting added to
the next block.

Adding tests to ensure the compressed CDXJ is working as expected now.

Fixes #1121

---------

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

Author: ikreymer

src/util/wacz.ts

@@ -326,6 +326,12 @@ export async function mergeCDXJ(
   indexesDir: string,
   zipped: boolean | null = null,
 ) {
+  // added as env vars as really just used for testing
+  const zipLinesPerBlock =
+    parseInt(process.env.ZIP_LINES_PER_BLOCK || "") || LINES_PER_BLOCK;
+  const zipCdxMinSize =
+    parseInt(process.env.ZIP_CDX_MIN_SIZE || "") || ZIP_CDX_MIN_SIZE;
+
   async function* readLinesFrom(stdout: Readable): AsyncGenerator<string> {
     for await (const line of readline.createInterface({ input: stdout })) {
       yield line + "\n";
@@ -335,6 +341,7 @@ export async function mergeCDXJ(
   async function* generateCompressed(
     reader: AsyncGenerator<string>,
     idxFile: Writable,
+    zipLinesPerBlock: number,
   ) {
     let offset = 0;
 
@@ -345,7 +352,6 @@ export async function mergeCDXJ(
     let cdxLines: string[] = [];
 
     let key = "";
-    let count = 0;
 
     idxFile.write(
       `!meta 0 ${JSON.stringify({
@@ -374,7 +380,6 @@ export async function mergeCDXJ(
 
       offset += length;
 
-      count = 1;
       key = "";
       cdxLines = [];
 
@@ -386,10 +391,11 @@ export async function mergeCDXJ(
         key = cdx.split(" {", 1)[0];
       }
 
-      if (++count === LINES_PER_BLOCK) {
+      cdxLines.push(cdx);
+
+      if (cdxLines.length === zipLinesPerBlock) {
         yield await finishChunk();
       }
-      cdxLines.push(cdx);
     }
 
     if (key) {
@@ -418,7 +424,7 @@ export async function mergeCDXJ(
     const tempCdxSize = await getDirSize(warcCdxDir);
 
     // if CDX size is at least this size, use compressed version
-    zipped = tempCdxSize >= ZIP_CDX_MIN_SIZE;
+    zipped = tempCdxSize >= zipCdxMinSize;
   }
 
   const proc = child_process.spawn("sort", cdxFiles, {
@@ -440,7 +446,13 @@ export async function mergeCDXJ(
     });
 
     await pipeline(
-      Readable.from(generateCompressed(readLinesFrom(proc.stdout), outputIdx)),
+      Readable.from(
+        generateCompressed(
+          readLinesFrom(proc.stdout),
+          outputIdx,
+          zipLinesPerBlock,
+        ),
+      ),
       output,
     );
 

tests/zipped-cdx-index.test.ts

@@ -0,0 +1,86 @@
+import { execSync } from "child_process";
+import fs from "fs";
+import { gunzipSync } from "zlib";
+
+const LINES_PER_BLOCK = 16;
+
+test("ensure basic crawl run with docker run passes", async () => {
+  execSync(
+    `docker run -e ZIP_LINES_PER_BLOCK=${LINES_PER_BLOCK} -e ZIP_CDX_MIN_SIZE=10000 -v $PWD/test-crawls:/crawls webrecorder/browsertrix-crawler crawl --url https://old.webrecorder.net/ --limit 6 --collection zipped-cdx-merge --generateWACZ`,
+  );
+});
+
+test("check that indexes/index.cdx.gz and indexes/index.idx exist", () => {
+  expect(
+    fs.existsSync("test-crawls/collections/zipped-cdx-merge/indexes/index.idx"),
+  ).toBe(true);
+
+  expect(
+    fs.existsSync(
+      "test-crawls/collections/zipped-cdx-merge/indexes/index.cdx.gz",
+    ),
+  ).toBe(true);
+});
+
+test("verify that index entries are as expected", async () => {
+  const indexBuff = fs.readFileSync(
+    "test-crawls/collections/zipped-cdx-merge/indexes/index.idx",
+    { encoding: "utf-8" },
+  );
+
+  const gzipBuff = new Uint8Array(
+    fs.readFileSync(
+      "test-crawls/collections/zipped-cdx-merge/indexes/index.cdx.gz",
+    ),
+  );
+
+  const blockLineCounts = [];
+
+  let allCdxLines: string[] = [];
+
+  for (const line of indexBuff.trim().split("\n")) {
+    if (line.startsWith("!meta 0")) {
+      continue;
+    }
+    // parse index line in format <surt> <ts> <json>, e.g.
+    // com,example)/ 20261226010203 {...}
+    const prefixIndex = line.indexOf(" ");
+    // surt key prefix
+    const prefix = line.slice(0, prefixIndex);
+    const tsIndex = line.indexOf(" ", prefixIndex + 1);
+    // timestamp
+    const timestamp = line.slice(prefixIndex + 1, tsIndex);
+
+    const { offset, length } = JSON.parse(line.slice(tsIndex + 1));
+
+    const cdxLineBuff = new TextDecoder().decode(
+      gunzipSync(gzipBuff.slice(offset, offset + length)),
+    );
+    const cdxLines = cdxLineBuff.trim().split("\n");
+
+    // surt and timestamp of first CDX line match index surt and timestamp
+    const parts = cdxLines[0].split(" ");
+    expect(parts[0]).toBe(prefix);
+    expect(parts[1]).toBe(timestamp);
+
+    allCdxLines = [...allCdxLines, ...cdxLines];
+
+    // add to line counts
+    blockLineCounts.push(cdxLines.length);
+  }
+
+  // each line count is equal to number of lines per block, except last one
+  for (let i = 0; i < blockLineCounts.length - 1; i++) {
+    expect(blockLineCounts[i]).toBe(LINES_PER_BLOCK);
+  }
+
+  const remainder = allCdxLines.length % LINES_PER_BLOCK;
+
+  // should have the remainder
+  expect(blockLineCounts[blockLineCounts.length - 1]).toBe(remainder);
+
+  // ensure full sort of all cdx lines
+  for (let i = 0; i < allCdxLines.length - 1; i++) {
+    expect(allCdxLines[i] <= allCdxLines[i + 1]).toBe(true);
+  }
+});