feat(tokenize): skip and record blobs whose tokenizer fails

[EllianCarlos]

A srcML crash, an empty output or a timeout on one blob stopped step 2 for the whole project. With this PR, blobExec drops that blob, records it, and the run continues.

This PR targets deploy/reblame-c100, the code the -C100 re-blame runs use. It is not based on master.

Commits

1. Skip and record. A failed blob is dropped from the rewritten trees, the same way a denylisted blob is dropped. The exit status is 0. Each dropped, denylisted or oversized blob gets one row in <work>/tokenize-skipped.tsv (sha path reason detail tokenizer), and step 2 prints a count for each reason. A timed-out blob gets a retry_blob row, and the next run tries it first. --strict-tokenize (or CREGIT_STRICT_TOKENIZE=1) restores the old stop on exit 4 and 6. Sharded mode is always strict.
2. Timeout retries, off by default. --max-retries N (or CREGIT_MAX_RETRIES) gives a timed-out blob up to N more attempts in the same run. Each retry waits for the load to drop and gets 3x the time budget. The default is 0, so the stall watchdog window stays at 30 min. With 3 retries the window grows to about 2 h 20 min.

Retry and blame

When a blob from an earlier run recovers, the run folds all of history again. All commit ids change, so the runner deletes blame/ and blame-c100-incoming/, and the project needs a full re-blame. To prevent that, step 2 does not retry earlier timeouts while a .blame file exists. It prints how many blobs stay dropped and how many .blame files a retry would delete. --retry-skipped (or CREGIT_RETRY_SKIPPED=1) forces the retry.

Tests

Suite	Passed	Failed
sbt -batch test	201	0
test_tokenize_gate.sh	90	0
test_ensure_artifacts.sh	42	0
test_retokenize_passthrough.sh	32	0
test_reblame_passthrough.sh	15	0
test_run_pipeline_workdir_guard.sh	75	0
prove (tokenize, tokenizeByBlobId, blameRepo, prettyPrint)	425	0
pytest generate_dataset	68	0

Not tested

• A real srcML crash in a full project run (the tests use a fake parser).
• A run through ctp.py.
• Sharded mode with retries (sharded mode stays strict).

[EllianCarlos]

Automated review (Claude Code, three axes: spec, standards, crop). I checked the main findings in the code before I posted it.

Verdict. The skip-and-record path does what the PR says in the three walkers, and the tests pass (I ran SkipFailedBlobSpec + MainOptionsSpec: 68 passed, 0 failed; test_tokenize_gate.sh in devenv shell: 90 passed, 0 failed). But the re-fold after a recovered timeout is not safe if the run stops part way: the recovered blob or the stale-blame cleanup can be lost for good. Strict mode also exits 0 while held blobs stay dropped. The changeset can lose about 120 lines with no loss of tested behaviour, and about 630 lines if the in-run retry moves to its own PR.

#	Severity	Where	Finding	Lines saved
1	High	Walker.scala 1446-1463, Main.scala 864-869	clearFold() and the refold marker come after the per-blob commits and after the walk. A stop in between omits the recovered blob for good, or keeps stale .blame files.	0
2	Medium	Walker.scala 1355-1366, 1225-1226	Strict mode plus --no-retry-timed-out (passed whenever blame exists) drops blobs and exits 0. The doc says skippedKeys is empty under strict.	0
3	Low	run_pipeline_process.sh 615-617, Walker.scala 1256	The warning says the next run retries a timeout. After step 7 it does not, unless --retry-skipped.	0
4	Crop	Walker.scala 947-959, 1104-1108, 1412-1417, 1243, 1227, 1276	Dead AtomicBoolean/AtomicReference wrappers, stale "callback" comments, dead "unknown" fallbacks, a never-read map value.	14
5	Crop	Walker.scala 1254-1268	One SKIPPED blob log line takes 15 source lines.	8
6	Crop	Main.scala 478-509	Four copies of the same numeric-option parse block; !v.isNaN does nothing.	13
7	Standards / crop	LoadGate.scala 35-46	onPoll is test-only, and its doc says the opposite of the caller.	3
8	Crop	SkipFailedBlobSpec.scala 386-422	Two identical tests, one subset of a loop test, repeated resolveStallWindow asserts.	27
9	Low / crop	test_tokenize_gate.sh 376-393	Label says factor 0 disables the retry (false). Two pairs of copy checks.	6
10	Crop	README.md 220-341, Main.scala 126 and 256-262, run_pipeline_process.sh 168-175	The worst-case figures (7820 s, 8420 s, "2 h 20 min", "30 min") are written in 4 places, and the README tells the re-fold and hold story twice. Keep one copy in the README.	about 25
11	Crop	Walker.scala 1286-1306, 1368-1387	The scaladocs of tokenize (21 lines) and retryTimedOutBlobs (20 lines) repeat the code branch by branch. About 8 lines each is enough.	about 25
12	Split candidate	commit 4d3eecb	The in-run retry is off by default (the decision to keep the 30-min watchdog). It needs LoadGate.scala (69), the retry loop in tokenize, retryBudget/longestBlobSeconds, resolveStallWindow, 4 options with help, the runner pass-through and validation, case 19, the spec section at 331-528, and the README retry text. Move it to a follow-up PR. Keep --no-retry-timed-out, --retry-skipped and the blame hold here: they protect blame from the cross-run retry of commit 1.	about 600

Total estimated lines saved: about 120 in this PR (items 4-11). With the split (item 12), about 630, because the split also removes most of items 6-10.

Other notes, not blocking:

• Sharded mode ignores CREGIT_MAX_RETRIES, CREGIT_TIMEOUT_RETRY_FACTOR and CREGIT_LOAD_LIMIT with no warning (run_pipeline_process.sh 1395-1398). The PR body lists this as not tested.
• test_tokenize_gate.sh fails 48 of 90 checks outside devenv shell. In the shell, all pass. This is the environment, not the PR.

[EllianCarlos]

[High, spec] The re-fold state is lost if the run stops before the walk ends.

Each recovered blob commits putBlob + deleteRetry at once (lines 1447-1450). clearFold() runs only after the loop (here), and Main writes --refold-marker only after the whole walk (Main.scala 864-869). Two failure windows follow:

1. The run stops after the first recovery commit and before clearFold(). The next loop rows can each take a full budget, so this window is long. The next run finds no retry_blob row and a full commit_map. It does not visit the old commits again. The recovered blob is then omitted from the tokenized repository for good, with no record.
2. The run stops during the re-fold walk (the watchdog, OOM, or a kill). The next run finds no retry_blob row, so refolded is false and no marker is written. The runner keeps blame/ and blame-c100-incoming/, which name commits that no longer exist. The runner comment at run_pipeline_process.sh 1468-1471 says that step 7 then keeps those stale .blame files silently.

Fix: reset the fold and write the marker before the first blob row. The runner acts on the marker only after step 2 exits 0 (tokenize_gate stops it first), so an early marker is safe.

@@ retryTimedOutBlobs, case other @@
                 inserter.flush()
+                if (recovered == 0) {
+                  // Before the first blob row, so that a stop at any later point
+                  // still re-folds and still drops the old outputs.
+                  onRefold()          // Main writes --refold-marker here
+                  mapping.clearFold()
+                  refolded = true
+                }
                 mapping.inTx {
@@ after the loop @@
     if (recovered > 0) {
-      mapping.clearFold()
-      refolded = true
       blobsRecovered.add(recovered)
@@ Main.scala 864-869 @@
-    // Written after the walk, so that it exists only if the re-fold is complete.
-    if (stats.refolded) refoldMarker.foreach { p => ... }
+    // Pass to the Walker instead: onRefold = () => refoldMarker.foreach(writeMarker)

Add a test that stops the walk after retryTimedOutBlobs (for example a stall) and checks that the next run keeps the blob and that the marker exists.

[EllianCarlos]

[Medium, spec] Strict mode exits 0 while held blobs stay dropped.

The runner passes --no-retry-timed-out whenever blame output exists, also with --strict-tokenize (run_pipeline_process.sh 1420-1433). Here the held keys go into skippedKeys, and no counter changes. So a strict run after a default run drops those blobs from every new tree and exits 0. The PR says that strict mode restores the old stop on exit 4 and 6, and report_skipped_blobs tells the operator to use strict mode "to require a clean run". The doc of skippedKeys (lines 1225-1226) also says "Empty under strictTokenize". That is false here and in retryTimedOutBlobs (lines 1418, 1435).

Count the held blobs in strict mode (or make the runner refuse strict mode together with a hold), and correct the doc:

Suggested change

	pending.foreach { key =>
	skippedKeys.put(key, BlobExec.Failure(BlobExec.Failure.Timeout, "retry pass off"))
	pendingRetry.add(key)
	}
	pending.foreach { key =>
	skippedKeys.put(key, BlobExec.Failure(BlobExec.Failure.Timeout, "retry pass off"))
	pendingRetry.add(key)
	}
	// Strict mode must not exit 0 while it keeps blobs dropped.
	if (strictTokenize) blobsTimedOut.add(pending.size.toLong)

[EllianCarlos]

[Low, spec] The warning says that the next run retries a timeout. That is false after step 7.

After the first blame, every later step-2 run gets --no-retry-timed-out (line 1433). Thus the next run does not retry the timeout unless the operator gives --retry-skipped. The same false claim is in the SKIPPED blob line of Walker.scala 1256.

Suggested change

	log " repository, not as raw source and not as empty tokenizations. A timeout"
	log " is tried again by the next step-2 run. To require a clean run, use"
	log " --strict-tokenize (or CREGIT_STRICT_TOKENIZE=1)."
	log " repository, not as raw source and not as empty tokenizations. The next"
	log " step-2 run tries a timeout again only if no blame output exists, or with"
	log " --retry-skipped. To require a clean run, use --strict-tokenize (or CREGIT_STRICT_TOKENIZE=1)."

[EllianCarlos]

[Crop, about 14 lines] Dead wrappers and stale comments in 3 places.

tokenize now returns failed: Option[Failure]. The code puts it in a new AtomicBoolean and a new AtomicReference, and reads them back on the same thread. No callback sets them. The comment "Set by either callback below" is no longer true. The same pattern is at lines 1104-1108 and 1412-1413. Also, dropFailedBlob is called only when failed.isDefined, so the "unknown" fallbacks at lines 1243 and 1416-1417 are dead. skippedKeys stores a Failure that no code reads (only containsKey/putIfAbsent), so a key set like oversizedKeys is enough. In noteTokenized (1276) the contains check repeats the filter that forget does.

Suggested change

	// Set by either callback below: both mean "this blob's tokenization is
	// unusable, so nothing about it may be persisted".
	val unusable = new AtomicBoolean(false)
	val outcome = BlobExec.run(
	bytes = bytes,
	origSha = task.origId.name,
	filename = task.filename,
	fullPath = task.fullPath,
	command = command,
	abortOnError = abortOnError,
	inserter = workerInserter,
	timeoutSeconds = blobTimeoutSeconds,
	onTimeout = () => { blobsTimedOut.increment(); unusable.set(true) },
	onParserCrash = () => { blobsParserCrashed.increment(); unusable.set(true) }
	)
	val (outcome, failed) = tokenize(bytes, task, workerInserter)
	val unusable = new AtomicBoolean(failed.isDefined)
	val failure = new AtomicReference[BlobExec.Failure](failed.orNull)
	val res = outcome match {
	case BlobExec.Outcome.Skip if unusable.get() && !strictTokenize =>
	// Dropped, exactly as a denylisted blob is: no id, so no tree entry, no
	// blob_map row and no dataset row. The original bytes are NOT put into
	// dst, because nothing refers to them.
	dropFailedBlob(task, failure.get())
	BlobResult.Oversized(task.origId)
	case BlobExec.Outcome.Skip if unusable.get() =>
	val (outcome, failed) = tokenize(bytes, task, workerInserter)
	val res = outcome match {
	case BlobExec.Outcome.Skip if failed.isDefined && !strictTokenize =>
	// Dropped, exactly as a denylisted blob is: no id, so no tree entry, no
	// blob_map row and no dataset row. The original bytes are NOT put into
	// dst, because nothing refers to them.
	dropFailedBlob(task, failed.get)
	BlobResult.Oversized(task.origId)
	case BlobExec.Outcome.Skip if failed.isDefined =>

The same change in the batch path:

-            // Set by either callback below: both mean "this blob's tokenization is
-            // unusable, so nothing about it may be persisted".
             val (outcome, failed) = tokenize(bytes, task, workerInserter)
-            val unusable = new AtomicBoolean(failed.isDefined)
-            val failure  = new AtomicReference[BlobExec.Failure](failed.orNull)
-            val dropped = unusable.get() && !strictTokenize
+            val dropped = failed.isDefined && !strictTokenize
-              case _ if dropped => dropFailedBlob(task, failure.get())
+              case _ if dropped => dropFailedBlob(task, failed.get)
-                if (!unusable.get()) noteTokenized(task)
+                if (failed.isEmpty) noteTokenized(task)
-            if (dropped) None else Some((task, outcome, unusable.get()))
+            if (dropped) None else Some((task, outcome, failed.isDefined))

[EllianCarlos]

[Crop, about 8 lines] One log line takes 15 source lines.

The README and --help already explain the drop. The line must give the facts and the SKIPPED blob prefix. This also removes the false claim that the next run retries the blob (see the comment on run_pipeline_process.sh 615).

Suggested change

	val next =
	if (f.reason == BlobExec.Failure.Timeout)
	"A timeout depends on load, so the next run over this memo tries this blob again."
	else
	"A parser crash or empty output is deterministic for this tokenizer, so a re-run " +
	"does not try it again " +
	"unless the tokenizer identity changes (--retokenize)."
	System.err.println(
	s"blobExec: SKIPPED blob: sha=$sha path=${task.fullPath} reason=${f.reason} " +
	s"detail=${f.detail}. The tokenizer output is not usable, so the blob is dropped from " +
	"the rewritten tree, as a denylisted blob is: it gives no blame and no dataset row, " +
	"and the file is not in the tokenized repository as raw source. The walk continues. " +
	next +
	skipLog.path.map(p => s" Recorded in $p.").getOrElse("")
	)
	val next =
	if (f.reason == BlobExec.Failure.Timeout) " A later run can try it again (retry_blob)."
	else " A re-run tries it again only if the tokenizer identity changes (--retokenize)."
	System.err.println(
	s"blobExec: SKIPPED blob: sha=$sha path=${task.fullPath} reason=${f.reason} " +
	s"detail=${f.detail}. It is dropped from the rewritten tree, as a denylisted blob is. " +
	"The walk continues." + next + skipLog.path.map(p => s" Recorded in $p.").getOrElse(""))

[EllianCarlos]

[Crop, about 13 lines] Four copies of the same parse block.

Each numeric option repeats stripPrefix, toIntOption.filter, match, println and sys.exit(1). The file already has parsePositiveSeconds (line 113) for the same job. Also, !v.isNaN at line 496 does nothing: NaN >= 0 is already false. One helper keeps the error texts the same:

+  private def intFlag(t: String, prefix: String, lo: Int, hi: Int, what: String): Int = {
+    val spec = t.stripPrefix(prefix)
+    spec.toIntOption.filter(n => n >= lo && n <= hi).getOrElse {
+      System.err.println(s"Error: ${prefix.dropRight(1)} must be $what [$spec]"); sys.exit(1)
+    }
+  }
...
       case t if t.startsWith("--max-retries=") =>
-        (8 lines)
+        maxRetries = intFlag(t, "--max-retries=", 0, 100, "a whole number from 0 to 100")
       case t if t.startsWith("--timeout-retry-factor=") =>
-        (8 lines)
+        timeoutRetryFactor = intFlag(t, "--timeout-retry-factor=", 0, Int.MaxValue, "a whole number >= 0")
       case t if t.startsWith("--load-limit=") =>
         val spec = t.stripPrefix("--load-limit=")
-        (7 lines)
+        loadLimit = spec.toDoubleOption.filter(_ >= 0).getOrElse {
+          System.err.println(s"Error: --load-limit must be a number >= 0 [$spec]"); sys.exit(1)
+        }
       case t if t.startsWith("--load-wait-max=") =>
-        (8 lines)
+        loadWaitMax = intFlag(t, "--load-wait-max=", 0, Int.MaxValue, "a whole number of seconds >= 0")

[EllianCarlos]

[Standards, crop about 3 lines] onPoll is test-only, and its doc says the opposite of what the caller does.

The doc says the caller can stamp progress during the wait. The only shipped caller passes _ => () (Walker.scala 1328), and the tokenize doc says that no progress is stamped during the waits, by design (Main sizes the stall window for it). Only SkipFailedBlobSpec 489-497 uses the parameter. Remove it, delete onPoll(load) at line 46, and let the tests check the returned seconds.

Suggested change

	/** Wait while the load is above the limit. `onPoll` is called at each poll with
	* the load, so the caller can stamp progress (the stall watchdog must not count
	* a deliberate wait as a stall). Returns the seconds waited. */
	def await(onPoll: Double => Unit): Long = {
	/** Wait while the load is above the limit. No progress is stamped: Main makes
	* the stall window larger than the longest wait. Returns the seconds waited. */
	def await(): Long = {

[EllianCarlos]

[Crop, about 27 lines] Four tests repeat other tests.

• 386-393 and 406-413 have the same body. maxRetries = 0 and Main.DefaultMaxRetries are the same value (line 396 asserts it).
• The resolveStallWindow asserts at 399-403 repeat lines 511 and 521.
• 415-422 is the serial (1, 1) case of the loop at 335-352, which asserts more.

One test keeps the default, the help text and the behaviour:

Suggested change

	test("--max-retries=0 gives no retry") {
	val fx = fixture("timeout-1")
	val stats = run(fx, "serial", maxRetries = 0)
	stats.blobsTimeoutRetriedInRun shouldEqual 0L
	stats.blobsSkipped shouldEqual 1L
	fx.callsForB shouldEqual 1
	tsvLines(fx).last.split("\t")(3) shouldEqual "timeout=1s"
	}
	test("the default is no retry, and the stall window then stays at 3 x --blob-timeout") {
	Main.DefaultMaxRetries shouldEqual 0
	Main.Usage should include("default 0: a timeout skips and records the blob")
	// With the default, a defaulted window stays at 1800s (30 min) for 600s blobs.
	Main.resolveStallWindow(600, Main.DefaultMaxRetries, Main.DefaultTimeoutRetryFactor,
	600, 1800, stallExplicit = false) shouldEqual Right(1800)
	// With an explicit --max-retries=3 it grows to 8420s (about 2 h 20 min).
	Main.resolveStallWindow(600, 3, Main.DefaultTimeoutRetryFactor,
	600, 1800, stallExplicit = false) shouldEqual Right(8420)
	}
	test("with the default, a timeout skips and records the blob at once") {
	val fx = fixture("timeout-1")
	val stats = run(fx, "serial", maxRetries = Main.DefaultMaxRetries)
	stats.blobsTimeoutRetriedInRun shouldEqual 0L
	stats.blobsSkipped shouldEqual 1L
	fx.callsForB shouldEqual 1
	tsvLines(fx).last.split("\t")(3) shouldEqual "timeout=1s"
	}
	test("an explicit --max-retries=3 still retries") {
	val fx = fixture("timeout-1")
	val stats = run(fx, "serial", maxRetries = 3)
	stats.blobsTimeoutRetriedInRun shouldEqual 1L
	stats.blobsTimeoutRecoveredInRun shouldEqual 1L
	stats.blobsSkipped shouldEqual 0L
	fx.callsForB shouldEqual 2
	}
	test("the default is no retry: a timeout skips and records the blob at once") {
	Main.DefaultMaxRetries shouldEqual 0
	Main.Usage should include("default 0: a timeout skips and records the blob")
	val fx = fixture("timeout-1")
	val stats = run(fx, "serial", maxRetries = Main.DefaultMaxRetries)
	stats.blobsTimeoutRetriedInRun shouldEqual 0L
	stats.blobsSkipped shouldEqual 1L
	fx.callsForB shouldEqual 1
	tsvLines(fx).last.split("\t")(3) shouldEqual "timeout=1s"
	}

[EllianCarlos]

[Low, crop about 6 lines] A wrong test label, and two pairs of copies.

The label at line 376 says "0 disables the retry". blobExec says the opposite: a factor of 0 or 1 gives the first budget (Walker.retryBudget, Main.scala 248, README 172). The retries still run. Also, 382-387 test the flag pass-through twice, and 388-393 test the CREGIT_MAX_RETRIES pass-through twice. Line 395 already covers the default of 0 (no flag is passed).

Suggested change

	grep -q -- "--timeout-retry-factor=0" "$STUB_ARGV"; check "the factor is passed through (0 disables the retry)" $?
	grep -q -- "--load-limit=8" "$STUB_ARGV"; check "the load limit is passed through" $?
	rm -f "$STUB_ARGV"
	OUT=$(STUB_RC=0 CREGIT_TIMEOUT_RETRY_FACTOR=5 run_step2 "$W")
	grep -q -- "--timeout-retry-factor=5" "$STUB_ARGV"; check "CREGIT_TIMEOUT_RETRY_FACTOR is honoured" $?
	rm -f "$STUB_ARGV"
	OUT=$(STUB_RC=0 run_step2 "$W" --max-retries 0)
	grep -q -- "--max-retries=0" "$STUB_ARGV"; check "--max-retries is passed through" $?
	rm -f "$STUB_ARGV"
	OUT=$(STUB_RC=0 run_step2 "$W" --max-retries 3)
	grep -q -- "--max-retries=3" "$STUB_ARGV"; check "an explicit --max-retries 3 is passed through" $?
	rm -f "$STUB_ARGV"
	OUT=$(STUB_RC=0 CREGIT_MAX_RETRIES=3 run_step2 "$W")
	grep -q -- "--max-retries=3" "$STUB_ARGV"; check "CREGIT_MAX_RETRIES=3 is passed through" $?
	rm -f "$STUB_ARGV"
	OUT=$(STUB_RC=0 CREGIT_MAX_RETRIES=5 run_step2 "$W")
	grep -q -- "--max-retries=5" "$STUB_ARGV"; check "CREGIT_MAX_RETRIES is honoured" $?
	rm -f "$STUB_ARGV"
	grep -q -- "--timeout-retry-factor=0" "$STUB_ARGV"; check "the factor is passed through (0 = the first budget)" $?
	grep -q -- "--load-limit=8" "$STUB_ARGV"; check "the load limit is passed through" $?
	rm -f "$STUB_ARGV"
	OUT=$(STUB_RC=0 CREGIT_TIMEOUT_RETRY_FACTOR=5 run_step2 "$W")
	grep -q -- "--timeout-retry-factor=5" "$STUB_ARGV"; check "CREGIT_TIMEOUT_RETRY_FACTOR is honoured" $?
	rm -f "$STUB_ARGV"
	OUT=$(STUB_RC=0 run_step2 "$W" --max-retries 3)
	grep -q -- "--max-retries=3" "$STUB_ARGV"; check "an explicit --max-retries 3 is passed through" $?
	rm -f "$STUB_ARGV"
	OUT=$(STUB_RC=0 CREGIT_MAX_RETRIES=5 run_step2 "$W")
	grep -q -- "--max-retries=5" "$STUB_ARGV"; check "CREGIT_MAX_RETRIES is honoured" $?
	rm -f "$STUB_ARGV"

[EllianCarlos]

Replaced by a shorter version with no new flags.