From 46cb0f61c9a5077a5b3ee83534cfb8cba0487116 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Sat, 25 Jul 2026 16:38:37 +0000
Subject: [PATCH 1/4] Run the integration suite in parallel
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The 120-test integration suite ran serially at ~26 s. That is tolerable for a
plain `make check` but not for the sanitizer builds, which re-run the whole
suite at 4-8x the cost (~103 s under TSAN, ~215 s under ASAN), nor for
`integration-valgrind` at ~7x.

Add `-j` to tests/run.py: `auto` (the default) is min(nproc, 8), `-j 1` keeps
the previous serial TextTestRunner verbatim. `make integration JOBS=…` plumbs
it through, and `integration-valgrind` picks up the same knob.

Measured on a 4-core box against an XFS scratch: 26.4 s -> 7.9 s (3.3x), with
`make check` going 35 s -> 13 s. Serial and parallel agree exactly (120 tests,
4 skipped), and five repeat runs - including two oversubscribed at -j 8 - were
stable.

Two details worth keeping:

  - Workers are processes, not threads. test_long_path chdir()s, and cwd is
    process-global, so threads would race it. Tests were already isolated by
    setUp's per-test mkdtemp and its per-test hashfile, so no test changes
    were needed.

  - Work is dispatched one test at a time rather than as fixed shards, so the
    slow outliers (autotune ~1.3 s, vacuum ~0.7 s) can't strand a worker at
    the end of the run.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01AxN8SYwbj24nAyAAsPRdm4
---
 CLAUDE.md    |  10 +++++
 Makefile     |  10 ++++-
 tests/run.py | 110 +++++++++++++++++++++++++++++++++++++++++++++------
 3 files changed, 116 insertions(+), 14 deletions(-)

diff --git a/CLAUDE.md b/CLAUDE.md
index 6a762e61aaab..d234798b650c 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -80,6 +80,16 @@ against a scratch tree and assert on the hashfile and on-disk sharing. Dedupe
 cases need a reflink fs (`DUPEREMOVE_TEST_DIR`, set by `devenv.sh`). Keep tests
 in `tests/`; no shell tests.
 
+- **The suite runs in parallel by default** — `tests/run.py -j` (`auto` =
+  `min(nproc, 8)`), i.e. `make integration JOBS=…`. ~3.3× on 4 cores (26.4→7.9s);
+  the real payoff is the sanitizer builds, which re-run all of it. Workers are
+  *processes*, not threads, because `test_long_path` `chdir()`s; work is handed
+  out one test at a time so the slow ones (autotune, vacuum) don't strand a
+  worker. **`JOBS=1` restores the old serial runner** — use it when a failure's
+  interleaved output is hard to read, or to check a suspected ordering
+  dependency. Tests are isolated by `setUp`'s per-test `mkdtemp` + hashfile;
+  anything sharing state outside that will flake in parallel.
+
 - **Never scan/benchmark out of `/tmp` — it's tmpfs**, not reflink-capable and
   rejected by `is_fs_supported()`, so a scan there stores **0 files silently**
   and dedupe is a no-op. Use real btrfs/xfs and verify a non-zero file count
diff --git a/Makefile b/Makefile
index b521bb695889..d1dd1f641da4 100644
--- a/Makefile
+++ b/Makefile
@@ -140,9 +140,15 @@ test:
 
 # End-to-end suite (Python stdlib unittest). Dedupe cases need a reflink fs;
 # override the scratch dir with DUPEREMOVE_TEST_DIR=/path.
+#
+# JOBS is tests/run.py's -j: worker processes, or `auto` for min(nproc, 8).
+# The suite is ~4x faster in parallel, which matters most for the sanitizer
+# builds (they re-run all of it at 4-8x the cost). JOBS=1 forces the serial
+# runner - use it when a failure's interleaved output is hard to read.
+JOBS ?= auto
 .PHONY: integration
 integration: oans
-	$(SANITIZE_RUN) DUPEREMOVE=./oans python3 tests/run.py
+	$(SANITIZE_RUN) DUPEREMOVE=./oans python3 tests/run.py -j $(JOBS)
 
 # Same end-to-end suite, but every oans invocation runs under valgrind memcheck
 # (via tests/valgrind-wrap.sh). Findings go to per-pid logs; a non-empty log
@@ -153,7 +159,7 @@ VGLOGDIR = $(CURDIR)/.vglogs
 integration-valgrind: oans
 	@command -v valgrind >/dev/null 2>&1 || { echo "valgrind not installed"; exit 1; }
 	rm -rf $(VGLOGDIR) && mkdir -p $(VGLOGDIR)
-	OANS_VG_LOGDIR=$(VGLOGDIR) DUPEREMOVE=tests/valgrind-wrap.sh python3 tests/run.py
+	OANS_VG_LOGDIR=$(VGLOGDIR) DUPEREMOVE=tests/valgrind-wrap.sh python3 tests/run.py -j $(JOBS)
 	@if find $(VGLOGDIR) -type f -size +0c | grep -q .; then \
 		echo "=== valgrind reported errors/leaks ==="; \
 		find $(VGLOGDIR) -type f -size +0c -exec cat {} +; \
diff --git a/tests/run.py b/tests/run.py
index c313339b4eca..03f6eb213066 100755
--- a/tests/run.py
+++ b/tests/run.py
@@ -2,22 +2,36 @@
 """Run the oans integration tests.
 
 Thin wrapper over unittest that prints a short banner (which binary, where the
-scratch lives, whether reflink works) and lets you filter tests by substring.
+scratch lives, whether reflink works), lets you filter tests by substring, and
+runs the suite across several worker processes.
 
 Usage:
   tests/run.py                     run everything
   tests/run.py hardlink dedupe     only tests whose id contains a given string
+  tests/run.py -j 8                run on 8 workers
+  tests/run.py -j 1                force the plain serial unittest runner
   DUPEREMOVE=/path tests/run.py            test a specific binary
   DUPEREMOVE_TEST_DIR=/mnt/btrfs tests/run.py   choose the scratch filesystem
 
-Equivalent to `python3 -m unittest discover -s tests/integration`, minus the
-banner and filtering. Exit status is non-zero if any test fails.
+Parallelism defaults to `auto` (min(nproc, 8)); `-j 1` restores the old serial
+runner verbatim. Every test already gets its own tempfile.mkdtemp() scratch with
+its own hashfile inside it, so tests don't collide. Workers are *processes*, not
+threads, because test_long_path chdir()s and cwd is process-global. Work is
+handed out one test at a time, so the slow tests (autotune, vacuum) don't strand
+a worker at the end.
+
+Exit status is non-zero if any test fails.
 """
 
+import argparse
+import concurrent.futures as futures
+import io
 import os
 import subprocess
 import sys
+import time
 import unittest
+from collections import Counter
 
 HERE = os.path.dirname(os.path.abspath(__file__))
 INTEGRATION_DIR = os.path.join(HERE, "integration")
@@ -25,6 +39,9 @@
 
 import harness  # noqa: E402  (needs the sys.path insert above)
 
+# Past this the scratch filesystem, not the CPU, is the limiting factor.
+MAX_AUTO_JOBS = 8
+
 
 def _matches(test_id, patterns):
     return not patterns or any(p in test_id for p in patterns)
@@ -58,8 +75,79 @@ def _tsan_note():
             "sets them.")
 
 
+def _jobs(value):
+    """Parse -j: a positive count, or 'auto' for min(nproc, MAX_AUTO_JOBS)."""
+    if value == "auto":
+        return min(os.cpu_count() or 1, MAX_AUTO_JOBS)
+    try:
+        jobs = int(value)
+    except ValueError:
+        raise argparse.ArgumentTypeError(f"not a number or 'auto': {value}")
+    if jobs < 1:
+        raise argparse.ArgumentTypeError("must be >= 1")
+    return jobs
+
+
+def _run_one(test_id):
+    """Run one test in this worker and return a picklable summary of it."""
+    suite = unittest.TestLoader().loadTestsFromName(test_id)
+    started = time.monotonic()
+    result = unittest.TextTestRunner(stream=io.StringIO(), verbosity=0).run(suite)
+    elapsed = time.monotonic() - started
+
+    if result.errors:
+        return "ERROR", elapsed, result.errors[0][1]
+    if result.failures:
+        return "FAIL", elapsed, result.failures[0][1]
+    if result.skipped:
+        return "skip", elapsed, result.skipped[0][1]
+    return "ok", elapsed, ""
+
+
+def _run_parallel(test_ids, jobs):
+    counts = Counter()
+    problems = []
+    started = time.monotonic()
+
+    with futures.ProcessPoolExecutor(max_workers=jobs) as pool:
+        pending = {pool.submit(_run_one, tid): tid for tid in test_ids}
+        for future in futures.as_completed(pending):
+            test_id = pending[future]
+            try:
+                status, elapsed, detail = future.result()
+            except Exception as exc:    # worker died: segfault, OOM, ...
+                status, elapsed, detail = "ERROR", 0.0, f"worker died: {exc}"
+            counts[status] += 1
+            print(f"{status:<5} {test_id} ({elapsed:.2f}s)", flush=True)
+            if status in ("FAIL", "ERROR"):
+                problems.append((status, test_id, detail))
+
+    for status, test_id, detail in problems:
+        print("=" * 70)
+        print(f"{status}: {test_id}")
+        print("-" * 70)
+        print(detail)
+
+    print("-" * 70)
+    print(f"Ran {sum(counts.values())} tests in "
+          f"{time.monotonic() - started:.2f}s on {jobs} workers\n")
+    if problems:
+        print(f"FAILED (failures={counts['FAIL']}, errors={counts['ERROR']}, "
+              f"skipped={counts['skip']})")
+        return 1
+    print(f"OK (skipped={counts['skip']})")
+    return 0
+
+
 def main(argv):
-    patterns = argv[1:]
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    parser.add_argument("-j", "--jobs", type=_jobs, default="auto",
+                        help="worker processes, or 'auto' (default: auto)")
+    parser.add_argument("patterns", nargs="*",
+                        help="only run tests whose id contains one of these")
+    # argparse applies `type` to a string default too, so this is already an int.
+    args = parser.parse_args(argv[1:])
+    jobs = args.jobs
 
     version = subprocess.run([harness.DUPEREMOVE, "--version"],
                              capture_output=True, text=True).stdout.strip()
@@ -69,6 +157,7 @@ def main(argv):
     print(f"  binary : {harness.DUPEREMOVE} ({version})")
     print(f"  scratch: {harness.TEST_ROOT} ({fstype})")
     print(f"  reflink: {'yes' if harness.REFLINK else 'no (dedupe tests will skip)'}")
+    print(f"  jobs   : {jobs}")
     tsan = _tsan_note()
     if tsan:
         print(f"  sanitize: {tsan}")
@@ -76,15 +165,12 @@ def main(argv):
 
     loader = unittest.TestLoader()
     discovered = loader.discover(start_dir=INTEGRATION_DIR, pattern="test_*.py")
+    tests = [t for t in _iter_tests(discovered) if _matches(t.id(), args.patterns)]
 
-    # Flatten and apply the substring filter.
-    suite = unittest.TestSuite()
-    for test in _iter_tests(discovered):
-        if _matches(test.id(), patterns):
-            suite.addTest(test)
-
-    result = unittest.TextTestRunner(verbosity=2).run(suite)
-    return 0 if result.wasSuccessful() else 1
+    if jobs == 1:
+        result = unittest.TextTestRunner(verbosity=2).run(unittest.TestSuite(tests))
+        return 0 if result.wasSuccessful() else 1
+    return _run_parallel([t.id() for t in tests], jobs)
 
 
 def _iter_tests(suite):

From 0c04e6303ea9c72c915b501fed92c201cd238306 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Sat, 25 Jul 2026 16:52:22 +0000
Subject: [PATCH 2/4] Simplify: one output path, and stop under-subscribing the
 pool

Follow-up to the parallel runner, from a review pass.

Reporting: the parallel path hand-rolled ~35 lines that unittest already
provides, which left two output formats to keep in step and silently dropped
information - only the first traceback per test, and nothing for
expectedFailure/unexpectedSuccess. Worker outcomes now replay into a real
TextTestResult, so unittest does the formatting and every category survives; a
test that fails in its body *and* errors in cleanup now shows both tracebacks.

With one formatter the `jobs == 1` special case dissolves: a one-worker pool is
already sequential, and the old branch meant TEST_JOBS=1 silently switched
execution model (in-process vs forked), so the mode you debugged in was not the
mode that failed.

Concurrency: `auto` was min(nproc, 8), which under-subscribes badly - the suite
blocks in subprocess.run(), sync() and autotune's cache drop, so total CPU is
flat across job counts. Measured on 4 cores over XFS: j=4 6.9s, j=8 4.4s,
j=12 3.6s, j=16 3.6s. Now 2 * nproc under the same ceiling, taking the suite
26.4s -> 4.3s (was 7.9s). The cap stays 8 rather than the ~12 plateau because a
worker under valgrind or ASAN costs far more than a plain oans; the comment now
says that instead of asserting a measurement I never made.

Also: pin the fork start method, which was load-bearing and undeclared (3.14
defaults Linux to forkserver, under which each worker would re-import harness
and re-probe reflink support); reuse harness.scratch_fstype() instead of a
second `stat -f` subprocess; rename JOBS to TEST_JOBS so it stops reading as
make's own -j; and cut the rationale from three copies to one.

Skipped from the review: a per-test `serial = True` opt-out (nothing needs it
yet) and longest-first scheduling (measured: no gain, discovery order already
happens to front-load the slow tests).

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01AxN8SYwbj24nAyAAsPRdm4
---
 CLAUDE.md       |  16 ++---
 Makefile        |  13 ++--
 tests/README.md |   6 ++
 tests/run.py    | 184 ++++++++++++++++++++++++++++++------------------
 4 files changed, 135 insertions(+), 84 deletions(-)

diff --git a/CLAUDE.md b/CLAUDE.md
index d234798b650c..4da709458e2a 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -80,15 +80,13 @@ against a scratch tree and assert on the hashfile and on-disk sharing. Dedupe
 cases need a reflink fs (`DUPEREMOVE_TEST_DIR`, set by `devenv.sh`). Keep tests
 in `tests/`; no shell tests.
 
-- **The suite runs in parallel by default** — `tests/run.py -j` (`auto` =
-  `min(nproc, 8)`), i.e. `make integration JOBS=…`. ~3.3× on 4 cores (26.4→7.9s);
-  the real payoff is the sanitizer builds, which re-run all of it. Workers are
-  *processes*, not threads, because `test_long_path` `chdir()`s; work is handed
-  out one test at a time so the slow ones (autotune, vacuum) don't strand a
-  worker. **`JOBS=1` restores the old serial runner** — use it when a failure's
-  interleaved output is hard to read, or to check a suspected ordering
-  dependency. Tests are isolated by `setUp`'s per-test `mkdtemp` + hashfile;
-  anything sharing state outside that will flake in parallel.
+- **The suite runs in parallel by default** (`make integration TEST_JOBS=…`,
+  `tests/run.py -j`; the why is in that file's docstring). 26.4→4.3s on 4 cores;
+  the real payoff is the sanitizer builds, which re-run all of it. **A new test
+  must not share state outside `setUp`'s per-test `mkdtemp` + hashfile** or it
+  will flake in parallel; `TEST_JOBS=1` is the sequential fallback for pinning
+  such a flake down. The suite is I/O-bound, so `auto` deliberately
+  over-subscribes (`2 × nproc`, capped) — don't "fix" it back to `nproc`.
 
 - **Never scan/benchmark out of `/tmp` — it's tmpfs**, not reflink-capable and
   rejected by `is_fs_supported()`, so a scan there stores **0 files silently**
diff --git a/Makefile b/Makefile
index d1dd1f641da4..e96769ac0916 100644
--- a/Makefile
+++ b/Makefile
@@ -141,14 +141,13 @@ test:
 # End-to-end suite (Python stdlib unittest). Dedupe cases need a reflink fs;
 # override the scratch dir with DUPEREMOVE_TEST_DIR=/path.
 #
-# JOBS is tests/run.py's -j: worker processes, or `auto` for min(nproc, 8).
-# The suite is ~4x faster in parallel, which matters most for the sanitizer
-# builds (they re-run all of it at 4-8x the cost). JOBS=1 forces the serial
-# runner - use it when a failure's interleaved output is hard to read.
-JOBS ?= auto
+# Worker processes for the suite - tests/run.py's -j, see `tests/run.py --help`.
+# Not make's own -j; TEST_JOBS=1 is the sequential fallback. The same value
+# sizes the valgrind leg below, where each worker is far heavier.
+TEST_JOBS ?= auto
 .PHONY: integration
 integration: oans
-	$(SANITIZE_RUN) DUPEREMOVE=./oans python3 tests/run.py -j $(JOBS)
+	$(SANITIZE_RUN) DUPEREMOVE=./oans python3 tests/run.py -j $(TEST_JOBS)
 
 # Same end-to-end suite, but every oans invocation runs under valgrind memcheck
 # (via tests/valgrind-wrap.sh). Findings go to per-pid logs; a non-empty log
@@ -159,7 +158,7 @@ VGLOGDIR = $(CURDIR)/.vglogs
 integration-valgrind: oans
 	@command -v valgrind >/dev/null 2>&1 || { echo "valgrind not installed"; exit 1; }
 	rm -rf $(VGLOGDIR) && mkdir -p $(VGLOGDIR)
-	OANS_VG_LOGDIR=$(VGLOGDIR) DUPEREMOVE=tests/valgrind-wrap.sh python3 tests/run.py -j $(JOBS)
+	OANS_VG_LOGDIR=$(VGLOGDIR) DUPEREMOVE=tests/valgrind-wrap.sh python3 tests/run.py -j $(TEST_JOBS)
 	@if find $(VGLOGDIR) -type f -size +0c | grep -q .; then \
 		echo "=== valgrind reported errors/leaks ==="; \
 		find $(VGLOGDIR) -type f -size +0c -exec cat {} +; \
diff --git a/tests/README.md b/tests/README.md
index 2c8da626e231..efbf48701d6d 100644
--- a/tests/README.md
+++ b/tests/README.md
@@ -21,11 +21,17 @@ Run both with `make check`.
 
 ```sh
 make integration                         # build oans and run the suite
+make integration TEST_JOBS=1             # ... sequentially (see -j below)
 python3 tests/run.py                      # run directly (binary must be built)
 python3 tests/run.py hardlink dedupe      # only tests whose id matches a pattern
+python3 tests/run.py -j 8                 # 8 workers ('auto' by default)
 python3 -m unittest discover -s tests/integration -v   # plain unittest, no banner
 ```
 
+The suite runs across worker processes by default, so tests must keep to the
+per-test scratch `setUp` hands them. `-j 1` runs them one at a time, which is
+where to start if a test only fails in company.
+
 Environment:
 
 - `DUPEREMOVE=/path/to/oans` — test a specific binary (defaults to the one
diff --git a/tests/run.py b/tests/run.py
index 03f6eb213066..e3e78c9bf4f4 100755
--- a/tests/run.py
+++ b/tests/run.py
@@ -3,35 +3,35 @@
 
 Thin wrapper over unittest that prints a short banner (which binary, where the
 scratch lives, whether reflink works), lets you filter tests by substring, and
-runs the suite across several worker processes.
+spreads the suite over worker processes.
 
 Usage:
   tests/run.py                     run everything
   tests/run.py hardlink dedupe     only tests whose id contains a given string
   tests/run.py -j 8                run on 8 workers
-  tests/run.py -j 1                force the plain serial unittest runner
+  tests/run.py -j 1                one worker, i.e. strictly sequential
   DUPEREMOVE=/path tests/run.py            test a specific binary
   DUPEREMOVE_TEST_DIR=/mnt/btrfs tests/run.py   choose the scratch filesystem
 
-Parallelism defaults to `auto` (min(nproc, 8)); `-j 1` restores the old serial
-runner verbatim. Every test already gets its own tempfile.mkdtemp() scratch with
-its own hashfile inside it, so tests don't collide. Workers are *processes*, not
-threads, because test_long_path chdir()s and cwd is process-global. Work is
-handed out one test at a time, so the slow tests (autotune, vacuum) don't strand
-a worker at the end.
+Workers are *processes*, not threads: test_long_path chdir()s and cwd is
+process-global. Tests need no cooperation to be split up - harness.setUp already
+gives each one its own mkdtemp scratch and its own hashfile inside it - but a
+test that reaches outside that will flake in parallel.
+
+Results are collected as they land and replayed into a real unittest result, so
+the output is unittest's own in every mode, `-j 1` included.
 
 Exit status is non-zero if any test fails.
 """
 
 import argparse
 import concurrent.futures as futures
-import io
+import multiprocessing
 import os
 import subprocess
 import sys
 import time
 import unittest
-from collections import Counter
 
 HERE = os.path.dirname(os.path.abspath(__file__))
 INTEGRATION_DIR = os.path.join(HERE, "integration")
@@ -39,9 +39,25 @@
 
 import harness  # noqa: E402  (needs the sys.path insert above)
 
-# Past this the scratch filesystem, not the CPU, is the limiting factor.
+# The suite is I/O-bound, not CPU-bound: workers sit in subprocess.run(), sync()
+# and autotune's cache drop, so total CPU is flat across job counts and nproc
+# alone under-subscribes badly. Measured on a 4-core box over XFS: j=4 6.9s,
+# j=8 4.4s, j=12 3.6s, j=16 3.6s. Hence 2x nproc, and a ceiling because a worker
+# is not always a plain oans - under valgrind or ASAN each one costs far more
+# memory, and the scratch filesystem is shared. 8 is a deliberate compromise
+# short of the ~12 plateau, not a measured optimum.
 MAX_AUTO_JOBS = 8
 
+# What a worker ships back per test, and the result method that replays it.
+# addUnexpectedSuccess takes no detail argument; the rest take one.
+_REPLAY = {
+    "failures": "addFailure",
+    "errors": "addError",
+    "skipped": "addSkip",
+    "expectedFailures": "addExpectedFailure",
+    "unexpectedSuccesses": "addUnexpectedSuccess",
+}
+
 
 def _matches(test_id, patterns):
     return not patterns or any(p in test_id for p in patterns)
@@ -76,88 +92,124 @@ def _tsan_note():
 
 
 def _jobs(value):
-    """Parse -j: a positive count, or 'auto' for min(nproc, MAX_AUTO_JOBS)."""
+    """Parse -j: a positive count, or 'auto'. argparse reports int()'s ValueError."""
     if value == "auto":
-        return min(os.cpu_count() or 1, MAX_AUTO_JOBS)
-    try:
-        jobs = int(value)
-    except ValueError:
-        raise argparse.ArgumentTypeError(f"not a number or 'auto': {value}")
+        return min(2 * (os.cpu_count() or 1), MAX_AUTO_JOBS)
+    jobs = int(value)
     if jobs < 1:
-        raise argparse.ArgumentTypeError("must be >= 1")
+        raise argparse.ArgumentTypeError("must be >= 1 or 'auto'")
     return jobs
 
 
+class _Stream:
+    """The writeln()-capable stream unittest's result object expects."""
+
+    def write(self, text):
+        sys.stdout.write(text)
+
+    def writeln(self, text=""):
+        sys.stdout.write(text + "\n")
+
+    def flush(self):
+        sys.stdout.flush()
+
+
+class _WorkerTest:
+    """Stands in for a test that ran in a worker, for display purposes only."""
+
+    def __init__(self, test_id):
+        self.test_id = test_id
+
+    def __str__(self):
+        return f"{self.test_id.rsplit('.', 1)[-1]} ({self.test_id})"
+
+    def shortDescription(self):
+        return None
+
+
+class _ReplayResult(unittest.TextTestResult):
+    """Collects worker outcomes; tracebacks arrive already formatted."""
+
+    def _exc_info_to_string(self, err, test):
+        return err
+
+
 def _run_one(test_id):
-    """Run one test in this worker and return a picklable summary of it."""
-    suite = unittest.TestLoader().loadTestsFromName(test_id)
+    """Run one test in this worker and return a picklable summary of it.
+
+    Dispatch is per test rather than per class, so setUpClass runs once per test
+    (120x rather than 33x). That is free today - harness's is a stat and a
+    makedirs - but a real class fixture would have to be dispatched as a unit.
+    """
+    result = unittest.TestResult()
     started = time.monotonic()
-    result = unittest.TextTestRunner(stream=io.StringIO(), verbosity=0).run(suite)
+    unittest.defaultTestLoader.loadTestsFromName(test_id).run(result)
     elapsed = time.monotonic() - started
 
-    if result.errors:
-        return "ERROR", elapsed, result.errors[0][1]
-    if result.failures:
-        return "FAIL", elapsed, result.failures[0][1]
-    if result.skipped:
-        return "skip", elapsed, result.skipped[0][1]
-    return "ok", elapsed, ""
+    outcomes = [(kind, detail)
+                for kind in ("failures", "errors", "skipped", "expectedFailures")
+                for _test, detail in getattr(result, kind)]
+    outcomes += [("unexpectedSuccesses", None)
+                 for _test in result.unexpectedSuccesses]
+    return elapsed, outcomes
 
 
-def _run_parallel(test_ids, jobs):
-    counts = Counter()
-    problems = []
+def _run_suite(test_ids, jobs, verbosity=2):
+    result = _ReplayResult(_Stream(), True, verbosity)
     started = time.monotonic()
 
-    with futures.ProcessPoolExecutor(max_workers=jobs) as pool:
+    # Pin fork: workers inherit the parent's already-imported test modules, so
+    # loadTestsFromName costs ~0.2ms. Python 3.14 defaults Linux to forkserver,
+    # under which every worker would re-import harness - whose module level
+    # probes reflink support and the scratch fstype - at ~0.15s each.
+    context = multiprocessing.get_context("fork")
+    with futures.ProcessPoolExecutor(max_workers=jobs, mp_context=context) as pool:
         pending = {pool.submit(_run_one, tid): tid for tid in test_ids}
         for future in futures.as_completed(pending):
-            test_id = pending[future]
+            test = _WorkerTest(pending[future])
+            result.startTest(test)
             try:
-                status, elapsed, detail = future.result()
-            except Exception as exc:    # worker died: segfault, OOM, ...
-                status, elapsed, detail = "ERROR", 0.0, f"worker died: {exc}"
-            counts[status] += 1
-            print(f"{status:<5} {test_id} ({elapsed:.2f}s)", flush=True)
-            if status in ("FAIL", "ERROR"):
-                problems.append((status, test_id, detail))
-
-    for status, test_id, detail in problems:
-        print("=" * 70)
-        print(f"{status}: {test_id}")
-        print("-" * 70)
-        print(detail)
-
-    print("-" * 70)
-    print(f"Ran {sum(counts.values())} tests in "
-          f"{time.monotonic() - started:.2f}s on {jobs} workers\n")
-    if problems:
-        print(f"FAILED (failures={counts['FAIL']}, errors={counts['ERROR']}, "
-              f"skipped={counts['skip']})")
-        return 1
-    print(f"OK (skipped={counts['skip']})")
-    return 0
+                elapsed, outcomes = future.result()
+            except Exception as exc:        # worker died: segfault, OOM, ...
+                outcomes = [("errors", f"worker died: {exc}\n")]
+            if not outcomes:
+                result.addSuccess(test)
+            for kind, detail in outcomes:
+                add = getattr(result, _REPLAY[kind])
+                add(test) if detail is None else add(test, detail)
+            result.stopTest(test)
+
+    result.printErrors()
+    result.stream.writeln(result.separator2)
+    result.stream.writeln(f"Ran {result.testsRun} tests in "
+                          f"{time.monotonic() - started:.2f}s on {jobs} workers")
+    result.stream.writeln()
+    if result.wasSuccessful():
+        result.stream.writeln(f"OK (skipped={len(result.skipped)})")
+        return 0
+    result.stream.writeln(f"FAILED (failures={len(result.failures)}, "
+                          f"errors={len(result.errors)}, "
+                          f"skipped={len(result.skipped)})")
+    return 1
 
 
 def main(argv):
-    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    parser = argparse.ArgumentParser(
+        description="Run the oans integration tests.",
+        formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__)
     parser.add_argument("-j", "--jobs", type=_jobs, default="auto",
                         help="worker processes, or 'auto' (default: auto)")
     parser.add_argument("patterns", nargs="*",
                         help="only run tests whose id contains one of these")
-    # argparse applies `type` to a string default too, so this is already an int.
     args = parser.parse_args(argv[1:])
-    jobs = args.jobs
 
     version = subprocess.run([harness.DUPEREMOVE, "--version"],
                              capture_output=True, text=True).stdout.strip()
-    fstype = subprocess.run(["stat", "-f", "-c", "%T", harness.TEST_ROOT],
-                            capture_output=True, text=True).stdout.strip()
     print("oans integration tests")
     print(f"  binary : {harness.DUPEREMOVE} ({version})")
-    print(f"  scratch: {harness.TEST_ROOT} ({fstype})")
+    print(f"  scratch: {harness.TEST_ROOT} ({harness.scratch_fstype()})")
     print(f"  reflink: {'yes' if harness.REFLINK else 'no (dedupe tests will skip)'}")
-    print(f"  jobs   : {jobs}")
+    print(f"  jobs   : {args.jobs}")
     tsan = _tsan_note()
     if tsan:
         print(f"  sanitize: {tsan}")
@@ -166,11 +218,7 @@ def main(argv):
     loader = unittest.TestLoader()
     discovered = loader.discover(start_dir=INTEGRATION_DIR, pattern="test_*.py")
     tests = [t for t in _iter_tests(discovered) if _matches(t.id(), args.patterns)]
-
-    if jobs == 1:
-        result = unittest.TextTestRunner(verbosity=2).run(unittest.TestSuite(tests))
-        return 0 if result.wasSuccessful() else 1
-    return _run_parallel([t.id() for t in tests], jobs)
+    return _run_suite([t.id() for t in tests], args.jobs)
 
 
 def _iter_tests(suite):

From 28f21e6fd20ccfa29c98c4284b60202766ac6bb8 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Sat, 25 Jul 2026 17:03:24 +0000
Subject: [PATCH 3/4] Fix two load-sensitive tests the parallel suite exposed

CI on btrfs failed test_extent_order_independent and test_streaming_dedupe,
and locally test_sparse_file_scans failed 4 runs in 12. Both are latent
dependencies on writeback that a serial suite never had to confront, not
breakage in the change - but they are the change's problem.

Physical extent layout. Several tests build a specific on-disk layout - the
fsync-forced extent boundary, or fiemap extent counts - and concurrent I/O
perturbs btrfs writeback enough that the setup does not get the layout it
intends. No amount of file isolation helps, so DuperemoveTest gains a `serial`
flag: tests/run.py holds those classes back and runs them one at a time after
the pool drains, keeping the parallel win for the other 110 tests. The four
fsync-boundary files are marked.

Delayed allocation. make_sparse handed a file straight to a scan asserting on
its extents, without ever fsyncing it. Under xfs delalloc those extents may not
be allocated yet, so oans maps none and correctly records none; running in
parallel pushed writeback further behind and turned a latent race into a 1-in-3
failure. The sparse-layout helpers now fsync, which is what the harness's own
sync() docstring already said this class of assertion needs. Fixed at the
helper rather than by serialising the test - the assertion is about oans, not
about writeback timing. 12 consecutive parallel runs clean, from 4/12 failing.

The serial flag is the mechanism the review pass proposed and I skipped as
speculative. CI found the case for it within the hour.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01AxN8SYwbj24nAyAAsPRdm4
---
 CLAUDE.md                                     |  7 ++++
 tests/README.md                               |  6 ++++
 tests/integration/harness.py                  | 29 ++++++++++++++--
 tests/integration/test_extent_dedupe.py       |  3 ++
 .../test_extent_order_independent.py          |  3 ++
 .../test_least_fragmented_target.py           |  3 ++
 tests/integration/test_streaming_dedupe.py    |  3 ++
 tests/run.py                                  | 33 +++++++++++++++----
 8 files changed, 78 insertions(+), 9 deletions(-)

diff --git a/CLAUDE.md b/CLAUDE.md
index 4da709458e2a..1bf530b02ae0 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -87,6 +87,13 @@ in `tests/`; no shell tests.
   will flake in parallel; `TEST_JOBS=1` is the sequential fallback for pinning
   such a flake down. The suite is I/O-bound, so `auto` deliberately
   over-subscribes (`2 × nproc`, capped) — don't "fix" it back to `nproc`.
+- **A test asserting on the *physical* extent layout must set `serial = True`**
+  (`DuperemoveTest.serial`), which holds it back to a one-at-a-time pass after
+  the pool drains. Per-test scratch isolation doesn't help here: the
+  fsync-forced-extent-boundary trick and fiemap counts depend on btrfs
+  writeback, which concurrent I/O perturbs — CI caught exactly this on btrfs
+  (`test_extent_order_independent`, `test_streaming_dedupe`) while xfs passed.
+  The four `fsync`-boundary files are already marked.
 
 - **Never scan/benchmark out of `/tmp` — it's tmpfs**, not reflink-capable and
   rejected by `is_fs_supported()`, so a scan there stores **0 files silently**
diff --git a/tests/README.md b/tests/README.md
index efbf48701d6d..3c07108fe991 100644
--- a/tests/README.md
+++ b/tests/README.md
@@ -58,6 +58,12 @@ Subclass `DuperemoveTest` (from `harness`) and add `test_*` methods. Each test
 gets a fresh scratch directory in `self.work` and a per-test hashfile in
 `self.hf`; both are cleaned up automatically.
 
+That isolation is what lets the suite run in parallel, so keep to it — a test
+reaching outside its own `self.work` will flake. If a test asserts on the
+*physical* extent layout (the fsync-forced extent boundary trick, or fiemap
+extent counts), set `serial = True` on the class: no amount of file isolation
+helps there, because concurrent I/O changes how the kernel lays extents out.
+
 ```python
 from harness import DuperemoveTest, requires_reflink
 
diff --git a/tests/integration/harness.py b/tests/integration/harness.py
index c95aa7cc1ace..c3d917cd2ef5 100644
--- a/tests/integration/harness.py
+++ b/tests/integration/harness.py
@@ -183,9 +183,30 @@ def scratch_fstype(directory=TEST_ROOT):
 # Base test case
 # --------------------------------------------------------------------------
 
+def _settle(f):
+    """fsync f so FIEMAP reports real extents rather than delayed allocation.
+
+    The sparse-layout helpers below build a file and hand it straight to a scan
+    that asserts on its extents. Without the fsync those extents may still be
+    unallocated when oans maps the file, and it correctly records none - which
+    surfaced as test_sparse_file_scans failing 4 runs in 12 once the suite
+    started running in parallel and writeback fell further behind.
+    """
+    f.flush()
+    os.fsync(f.fileno())
+
+
 class DuperemoveTest(unittest.TestCase):
     """Base class: a fresh scratch dir + hashfile per test, plus helpers."""
 
+    # Set True on a class whose assertions depend on the *physical* extent
+    # layout the kernel happens to produce - the fsync-forced extent boundary
+    # trick, or fiemap counts. Concurrent I/O perturbs btrfs writeback enough
+    # that the layout the setup intends is not the one it gets, so tests/run.py
+    # holds these back and runs them one at a time after the pool drains.
+    # Per-test scratch isolation is *not* what this is for; that already works.
+    serial = False
+
     @classmethod
     def setUpClass(cls):
         if not (os.path.isfile(DUPEREMOVE) and os.access(DUPEREMOVE, os.X_OK)):
@@ -378,12 +399,16 @@ def make_sparse(self, relpath, head, hole, tail):
             f.write(head)
             f.seek(len(head) + hole)
             f.write(tail)
+            _settle(f)
         return p
 
     def make_trailing_hole(self, relpath, data, size):
         """Write `data`, then extend the file to `size` so it ends in a hole."""
-        p = self.write(relpath, data)
-        os.truncate(p, size)
+        p = self.path(relpath)
+        with open(p, "wb") as f:
+            f.write(data)
+            f.truncate(size)
+            _settle(f)
         return p
 
     def reflink(self, src_rel, dst_rel):
diff --git a/tests/integration/test_extent_dedupe.py b/tests/integration/test_extent_dedupe.py
index 78abeaf70bdb..5ccc7355750f 100644
--- a/tests/integration/test_extent_dedupe.py
+++ b/tests/integration/test_extent_dedupe.py
@@ -23,6 +23,9 @@
 
 @requires_btrfs
 class ExtentDedupeTest(DuperemoveTest):
+    # Extent-layout sensitive: see DuperemoveTest.serial.
+    serial = True
+
     def _mkfile(self, rel, head, tail):
         """head, an fsync to force an extent boundary, then the shared tail."""
         p = self.path(rel)
diff --git a/tests/integration/test_extent_order_independent.py b/tests/integration/test_extent_order_independent.py
index 27f51712e827..c40e09a29661 100644
--- a/tests/integration/test_extent_order_independent.py
+++ b/tests/integration/test_extent_order_independent.py
@@ -20,6 +20,9 @@
 
 @requires_btrfs
 class ExtentOrderIndependentTest(DuperemoveTest):
+    # Extent-layout sensitive: see DuperemoveTest.serial.
+    serial = True
+
     def _mkfile(self, rel, head, tail):
         """head, an fsync to force an extent boundary, then the shared tail."""
         p = self.path(rel)
diff --git a/tests/integration/test_least_fragmented_target.py b/tests/integration/test_least_fragmented_target.py
index 027141e015f8..8ff6b43c0a8e 100644
--- a/tests/integration/test_least_fragmented_target.py
+++ b/tests/integration/test_least_fragmented_target.py
@@ -13,6 +13,9 @@
 
 @requires_btrfs
 class LeastFragmentedTargetTest(DuperemoveTest):
+    # Extent-layout sensitive: see DuperemoveTest.serial.
+    serial = True
+
     def _fragment(self, rel, content):
         """Write content, then rewrite alternate 4K blocks in place (COW) so the
         file ends up split across many physical extents."""
diff --git a/tests/integration/test_streaming_dedupe.py b/tests/integration/test_streaming_dedupe.py
index af7a0298dc5e..1bb6384728b4 100644
--- a/tests/integration/test_streaming_dedupe.py
+++ b/tests/integration/test_streaming_dedupe.py
@@ -18,6 +18,9 @@
 
 @requires_reflink
 class StreamingDedupeTest(DuperemoveTest):
+    # Extent-layout sensitive: see DuperemoveTest.serial.
+    serial = True
+
     # tiny passes + small batchsize => work spans many generation batches, so
     # the producer/watermark and cross-batch anchor paths are all exercised.
     ENV = {"DUPEREMOVE_FILES_PER_PASS": "8"}
diff --git a/tests/run.py b/tests/run.py
index e3e78c9bf4f4..e7af220603e7 100755
--- a/tests/run.py
+++ b/tests/run.py
@@ -16,7 +16,10 @@
 Workers are *processes*, not threads: test_long_path chdir()s and cwd is
 process-global. Tests need no cooperation to be split up - harness.setUp already
 gives each one its own mkdtemp scratch and its own hashfile inside it - but a
-test that reaches outside that will flake in parallel.
+test that reaches outside that will flake in parallel. The exception is tests
+asserting on the *physical* extent layout, which concurrent I/O perturbs no
+matter how isolated their files are; those set DuperemoveTest.serial and run one
+at a time once the pool has drained.
 
 Results are collected as they land and replayed into a real unittest result, so
 the output is unittest's own in every mode, `-j 1` included.
@@ -154,10 +157,8 @@ def _run_one(test_id):
     return elapsed, outcomes
 
 
-def _run_suite(test_ids, jobs, verbosity=2):
-    result = _ReplayResult(_Stream(), True, verbosity)
-    started = time.monotonic()
-
+def _dispatch(test_ids, jobs, result):
+    """Run test_ids across `jobs` workers, recording outcomes into `result`."""
     # Pin fork: workers inherit the parent's already-imported test modules, so
     # loadTestsFromName costs ~0.2ms. Python 3.14 defaults Linux to forkserver,
     # under which every worker would re-import harness - whose module level
@@ -179,10 +180,23 @@ def _run_suite(test_ids, jobs, verbosity=2):
                 add(test) if detail is None else add(test, detail)
             result.stopTest(test)
 
+
+def _run_suite(parallel_ids, serial_ids, jobs, verbosity=2):
+    result = _ReplayResult(_Stream(), True, verbosity)
+    started = time.monotonic()
+
+    _dispatch(parallel_ids, jobs, result)
+    # DuperemoveTest.serial tests want the box to themselves: their setup builds
+    # a specific physical extent layout, which concurrent I/O disturbs. Held to
+    # the end so they don't serialise the rest of the run either.
+    if serial_ids:
+        _dispatch(serial_ids, 1, result)
+
     result.printErrors()
+    tail = f" (+{len(serial_ids)} serial)" if serial_ids else ""
     result.stream.writeln(result.separator2)
     result.stream.writeln(f"Ran {result.testsRun} tests in "
-                          f"{time.monotonic() - started:.2f}s on {jobs} workers")
+                          f"{time.monotonic() - started:.2f}s on {jobs} workers{tail}")
     result.stream.writeln()
     if result.wasSuccessful():
         result.stream.writeln(f"OK (skipped={len(result.skipped)})")
@@ -218,7 +232,12 @@ def main(argv):
     loader = unittest.TestLoader()
     discovered = loader.discover(start_dir=INTEGRATION_DIR, pattern="test_*.py")
     tests = [t for t in _iter_tests(discovered) if _matches(t.id(), args.patterns)]
-    return _run_suite([t.id() for t in tests], args.jobs)
+    serial = [t.id() for t in tests if getattr(t, "serial", False)]
+    parallel = [t.id() for t in tests if not getattr(t, "serial", False)]
+    if serial:
+        print(f"  serial : {len(serial)} extent-layout tests held to the end\n",
+              flush=True)
+    return _run_suite(parallel, serial, args.jobs)
 
 
 def _iter_tests(suite):

From 68fd34296d429f283f73b0a20e15b5ce55b7050e Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Sat, 25 Jul 2026 17:53:17 +0000
Subject: [PATCH 4/4] Settle the scratch filesystem before each oans run

test_hardlink_pair_does_not_empty_hashfile saw 381 of 401 extents under the
TSAN build's load - the same delayed-allocation problem as the sparse tests,
just reached through mkrand/write rather than make_sparse. Twenty files had
not been written back when oans mapped them, so it recorded no extents for
them, correctly, and the test read the shortfall as a scanner bug.

Fixing it per file does work and costs far too much: fsync()ing every created
file is a journal commit apiece, and took the suite from 5.4s to 18s - most of
the parallel win, to make a guarantee no test needs at that granularity. What
the tests actually need is that the tree is on disk before oans walks it, which
is one syncfs() of the scratch at each oans invocation. That measures at 5.0s,
free against the 5.4s baseline.

So the per-helper fsyncs added with the sparse fix are gone again, replaced by
the single settle in dm(), which covers every test rather than the ones that
had already been caught flaking.

12 consecutive runs clean on the normal build, 5 on the TSAN build in parallel
with zero race reports.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01AxN8SYwbj24nAyAAsPRdm4
---
 tests/integration/harness.py | 34 ++++++++++++++++++++++------------
 1 file changed, 22 insertions(+), 12 deletions(-)

diff --git a/tests/integration/harness.py b/tests/integration/harness.py
index c3d917cd2ef5..1193f9feffe7 100644
--- a/tests/integration/harness.py
+++ b/tests/integration/harness.py
@@ -183,17 +183,28 @@ def scratch_fstype(directory=TEST_ROOT):
 # Base test case
 # --------------------------------------------------------------------------
 
-def _settle(f):
-    """fsync f so FIEMAP reports real extents rather than delayed allocation.
-
-    The sparse-layout helpers below build a file and hand it straight to a scan
-    that asserts on its extents. Without the fsync those extents may still be
-    unallocated when oans maps the file, and it correctly records none - which
-    surfaced as test_sparse_file_scans failing 4 runs in 12 once the suite
-    started running in parallel and writeback fell further behind.
+_libc = ctypes.CDLL("libc.so.6", use_errno=True)
+
+
+def _settle_scratch():
+    """syncfs() the scratch filesystem so FIEMAP sees real extents.
+
+    A test builds files and hands them straight to a scan that asserts on their
+    extents. Under delayed allocation those extents may not exist yet when oans
+    maps the file - it correctly records none, and the test reads the shortfall
+    as a scanner bug. Running the suite in parallel pushes writeback far enough
+    behind to lose this routinely: test_sparse_file_scans failed 4 runs in 12,
+    and test_hardlink_pair_does_not_empty_hashfile saw 381 of 401 extents.
+
+    Once per oans invocation, not once per file: fsync()ing each created file
+    costs a journal commit apiece and took the suite from 5.4s to 18s, where one
+    syncfs() of the whole scratch is a single syscall for the same guarantee.
     """
-    f.flush()
-    os.fsync(f.fileno())
+    fd = os.open(TEST_ROOT, os.O_RDONLY)
+    try:
+        _libc.syncfs(fd)
+    finally:
+        os.close(fd)
 
 
 class DuperemoveTest(unittest.TestCase):
@@ -232,6 +243,7 @@ def dm(self, *args, hashfile=True, stdin=None, env=None, quiet=True):
         Runs with -q by default (terse output); pass quiet=False to get the
         full human summary block (e.g. to assert on the 'Reclaimed' line).
         """
+        _settle_scratch()   # the tree must be on disk before oans maps it
         cmd = [DUPEREMOVE, "--io-threads=4"]
         if quiet:
             cmd.insert(1, "-q")
@@ -399,7 +411,6 @@ def make_sparse(self, relpath, head, hole, tail):
             f.write(head)
             f.seek(len(head) + hole)
             f.write(tail)
-            _settle(f)
         return p
 
     def make_trailing_hole(self, relpath, data, size):
@@ -408,7 +419,6 @@ def make_trailing_hole(self, relpath, data, size):
         with open(p, "wb") as f:
             f.write(data)
             f.truncate(size)
-            _settle(f)
         return p
 
     def reflink(self, src_rel, dst_rel):