| 1 | """Tiny-scale wire tests for the fleet-scale contention modes.
|
| 2 |
|
| 3 | Every new mode runs at --workers 2 scale against local git fixtures (worktrees
|
| 4 | or a local bare remote), asserting rows appear, pass the contention row
|
| 5 | contract, and carry the metrics each mode promises (queue_wait_ms,
|
| 6 | attempts_to_land, poll latencies, sentinel lost-update accounting). The
|
| 7 | incremental .partial.jsonl flusher is exercised directly and via the CLI.
|
| 8 | """
|
| 9 |
|
| 10 | from __future__ import annotations
|
| 11 |
|
| 12 | import json
|
| 13 | import os
|
| 14 | import shutil
|
| 15 | import subprocess
|
| 16 | import sys
|
| 17 | import tempfile
|
| 18 | import unittest
|
| 19 | from pathlib import Path
|
| 20 |
|
| 21 | sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
|
| 22 |
|
| 23 | import parallel_contention as lane
|
| 24 | from oakbench.results import ResultsStore
|
| 25 | from oakbench.rows import validate_rows
|
| 26 | from oakbench.subjects import Subject
|
| 27 |
|
| 28 | GIT = shutil.which("git")
|
| 29 | ROOT = Path(__file__).resolve().parents[1]
|
| 30 |
|
| 31 |
|
| 32 | def git_subject() -> Subject:
|
| 33 | return Subject(name="git", kind="git", label="Git", bin=Path(GIT))
|
| 34 |
|
| 35 |
|
| 36 | def make_metadata() -> dict:
|
| 37 | return {"bench_id": "test", "profile": "parallel-contention", "timestamp_utc": "test"}
|
| 38 |
|
| 39 |
|
| 40 | @unittest.skipIf(GIT is None, "git not installed")
|
| 41 | class FleetScenarioTestBase(unittest.TestCase):
|
| 42 | def setUp(self) -> None:
|
| 43 | self.tmp = Path(tempfile.mkdtemp(prefix="oakbench-fleet-test-"))
|
| 44 | self.fixture = self.tmp / "fixture"
|
| 45 | lane.make_fixture(self.fixture)
|
| 46 |
|
| 47 | def tearDown(self) -> None:
|
| 48 | shutil.rmtree(self.tmp, ignore_errors=True)
|
| 49 |
|
| 50 |
|
| 51 | class MergeTrainTest(FleetScenarioTestBase):
|
| 52 | def test_two_worker_train_lands_and_records_queue_waits(self) -> None:
|
| 53 | rows = lane.run_merge_train(
|
| 54 | git_subject(), self.fixture, workers=2, commits=1,
|
| 55 | run_root=self.tmp / "runs", run_index=0, metadata=make_metadata(),
|
| 56 | )
|
| 57 | self.assertEqual(validate_rows(rows, "contention"), [])
|
| 58 |
|
| 59 | worker_rows = [row for row in rows if row["operation"] == "parallel.worker"]
|
| 60 | self.assertEqual(len(worker_rows), 2)
|
| 61 | for row in worker_rows:
|
| 62 | self.assertEqual(row["scenario"], "contention_merge_train_w2")
|
| 63 | self.assertIsNotNone(row["queue_wait_ms"])
|
| 64 | self.assertGreaterEqual(row["queue_wait_ms"], 0.0)
|
| 65 | self.assertIsNotNone(row["merge_ms"])
|
| 66 | self.assertTrue(row["landed"])
|
| 67 |
|
| 68 | total = next(row for row in rows if row["operation"] == "parallel.total")
|
| 69 | self.assertEqual(total["returncode"], 0)
|
| 70 | self.assertEqual(total["merges_clean"], 2)
|
| 71 | self.assertEqual(total["merge_conflicts"], 0)
|
| 72 | self.assertIsNotNone(total["merges_per_min"])
|
| 73 | self.assertGreater(total["merges_per_min"], 0)
|
| 74 | self.assertIn("queue_wait_tail", total)
|
| 75 | metrics = total["parallel_metrics"]
|
| 76 | self.assertEqual(metrics["lost_updates_detected"], 0)
|
| 77 | self.assertTrue(metrics["integrity_check_passed"])
|
| 78 |
|
| 79 | def test_oak_subject_emits_structured_skip(self) -> None:
|
| 80 | oak = Subject(name="oak_installed", kind="oak", label="oak", bin=Path("/usr/bin/false"))
|
| 81 | rows = lane.run_merge_train(
|
| 82 | oak, self.fixture, workers=2, commits=1,
|
| 83 | run_root=self.tmp / "runs", run_index=0, metadata=make_metadata(),
|
| 84 | )
|
| 85 | self.assertEqual(len(rows), 1)
|
| 86 | self.assertEqual(rows[0]["returncode"], lane.SKIP_RETURNCODE)
|
| 87 | self.assertTrue(rows[0]["skipped"])
|
| 88 | self.assertIn("merge_train", rows[0]["skip_reason"])
|
| 89 | self.assertEqual(validate_rows(rows, "contention"), [])
|
| 90 |
|
| 91 |
|
| 92 | class RebaseStormTest(FleetScenarioTestBase):
|
| 93 | def test_two_workers_record_attempts_to_land(self) -> None:
|
| 94 | rows = lane.run_rebase_storm(
|
| 95 | git_subject(), self.fixture, workers=2, commits=1,
|
| 96 | run_root=self.tmp / "runs", run_index=0, metadata=make_metadata(),
|
| 97 | )
|
| 98 | self.assertEqual(validate_rows(rows, "contention"), [])
|
| 99 |
|
| 100 | worker_rows = [row for row in rows if row["operation"] == "parallel.worker"]
|
| 101 | self.assertEqual(len(worker_rows), 2)
|
| 102 | for row in worker_rows:
|
| 103 | self.assertEqual(row["scenario"], "contention_rebase_storm_w2")
|
| 104 | self.assertEqual(row["lands_succeeded"], 1)
|
| 105 | self.assertEqual(len(row["attempts_to_land"]), 1)
|
| 106 | self.assertGreaterEqual(row["attempts_to_land"][0], 1)
|
| 107 | self.assertEqual(row["remote_transport"], "local_file")
|
| 108 |
|
| 109 | total = next(row for row in rows if row["operation"] == "parallel.total")
|
| 110 | self.assertEqual(total["returncode"], 0)
|
| 111 | histogram = total["attempts_to_land_histogram"]
|
| 112 | self.assertEqual(sum(histogram.values()), 2)
|
| 113 | self.assertEqual(total["parallel_metrics"]["lost_updates_detected"], 0)
|
| 114 | self.assertTrue(total["parallel_metrics"]["integrity_check_passed"])
|
| 115 |
|
| 116 |
|
| 117 | class MixedFleetTest(FleetScenarioTestBase):
|
| 118 | def test_readers_poll_while_writers_commit(self) -> None:
|
| 119 | rows = lane.run_mixed_fleet(
|
| 120 | git_subject(), self.fixture, workers=2, commits=2,
|
| 121 | run_root=self.tmp / "runs", run_index=0, metadata=make_metadata(),
|
| 122 | )
|
| 123 | self.assertEqual(validate_rows(rows, "contention"), [])
|
| 124 |
|
| 125 | readers = [row for row in rows if row["operation"] == "parallel.reader"]
|
| 126 | writers = [row for row in rows if row["operation"] == "parallel.worker"]
|
| 127 | self.assertEqual(len(readers), 1)
|
| 128 | self.assertEqual(len(writers), 1)
|
| 129 | reader = readers[0]
|
| 130 | self.assertEqual(reader["scenario"], "contention_mixed_fleet_w2")
|
| 131 | self.assertGreaterEqual(reader["polls_total"], 1)
|
| 132 | self.assertEqual(reader["poll_failures"], 0)
|
| 133 | self.assertEqual(len(reader["poll_latency_samples_ms"]), reader["polls_total"])
|
| 134 | self.assertIn("p50_ms", reader["poll_latency_tail"])
|
| 135 |
|
| 136 | total = next(row for row in rows if row["operation"] == "parallel.total")
|
| 137 | self.assertEqual(total["readers"], 1)
|
| 138 | self.assertEqual(total["writers"], 1)
|
| 139 | self.assertEqual(total["poll_failures_total"], 0)
|
| 140 | self.assertTrue(total["parallel_metrics"]["integrity_check_passed"])
|
| 141 | self.assertEqual(total["returncode"], 0)
|
| 142 |
|
| 143 |
|
| 144 | class ConflictStormTest(FleetScenarioTestBase):
|
| 145 | def test_overlapping_appends_lose_nothing_and_account_for_sentinels(self) -> None:
|
| 146 | workers, commits = 2, 2
|
| 147 | rows = lane.run_conflict_storm(
|
| 148 | git_subject(), self.fixture, workers=workers, commits=commits,
|
| 149 | run_root=self.tmp / "runs", run_index=0, metadata=make_metadata(),
|
| 150 | )
|
| 151 | self.assertEqual(validate_rows(rows, "contention"), [])
|
| 152 |
|
| 153 | worker_rows = [row for row in rows if row["operation"] == "parallel.worker"]
|
| 154 | self.assertEqual(len(worker_rows), workers)
|
| 155 | for row in worker_rows:
|
| 156 | self.assertEqual(row["scenario"], "contention_conflict_storm_w2")
|
| 157 | self.assertEqual(row["sentinels_expected"], commits)
|
| 158 | # Serial-safe appends: every sentinel must survive into HEAD.
|
| 159 | self.assertEqual(row["sentinels_lost"], 0)
|
| 160 |
|
| 161 | total = next(row for row in rows if row["operation"] == "parallel.total")
|
| 162 | accounting = total["sentinel_accounting"]
|
| 163 | self.assertEqual(accounting["sentinels_expected"], workers * commits)
|
| 164 | self.assertEqual(accounting["sentinels_surviving"], workers * commits)
|
| 165 | self.assertEqual(accounting["lost_updates_detected"], 0)
|
| 166 | self.assertEqual(accounting["lost_updates_by_worker"], {})
|
| 167 | metrics = total["parallel_metrics"]
|
| 168 | self.assertEqual(metrics["overlapping_files_count"], lane.CONFLICT_STORM_FILES)
|
| 169 | self.assertEqual(metrics["lost_updates_detected"], 0)
|
| 170 | self.assertTrue(metrics["integrity_check_passed"])
|
| 171 |
|
| 172 |
|
| 173 | class ClonePushStormTest(FleetScenarioTestBase):
|
| 174 | def test_concurrent_clone_commit_push_against_one_local_remote(self) -> None:
|
| 175 | env_backup = os.environ.pop("GIT_BENCH_REMOTE", None)
|
| 176 | try:
|
| 177 | rows = lane.run_clone_push_storm(
|
| 178 | git_subject(), self.fixture, workers=2, commits=1,
|
| 179 | run_root=self.tmp / "runs", run_index=0, metadata=make_metadata(),
|
| 180 | )
|
| 181 | finally:
|
| 182 | if env_backup is not None:
|
| 183 | os.environ["GIT_BENCH_REMOTE"] = env_backup
|
| 184 | self.assertEqual(validate_rows(rows, "contention"), [])
|
| 185 | worker_rows = [row for row in rows if row["operation"] == "parallel.worker"]
|
| 186 | self.assertEqual(len(worker_rows), 2)
|
| 187 | for row in worker_rows:
|
| 188 | self.assertIsNotNone(row["clone_ms"])
|
| 189 | self.assertIsNotNone(row["push_ms"])
|
| 190 | self.assertEqual(row["remote_transport"], "local_file")
|
| 191 | # Measured completed-op counts on every worker row (ADR-0002:
|
| 192 | # observed integers, never null/absent), so fleet_report can tell
|
| 193 | # a starved worker (0) from an unmeasured one.
|
| 194 | self.assertIsInstance(row["pushes_succeeded"], int)
|
| 195 | self.assertIsInstance(row["clones_succeeded"], int)
|
| 196 | self.assertEqual(row["pushes_succeeded"], 1)
|
| 197 | self.assertEqual(row["clones_succeeded"], 1)
|
| 198 | total = next(row for row in rows if row["operation"] == "parallel.total")
|
| 199 | self.assertEqual(total["pushes_succeeded"], 2)
|
| 200 | self.assertEqual(total["returncode"], 0)
|
| 201 |
|
| 202 |
|
| 203 | class LongDivergenceTest(FleetScenarioTestBase):
|
| 204 | def test_k10_scenario_name_and_merge_cost_fields(self) -> None:
|
| 205 | rows = lane.run_long_divergence(
|
| 206 | git_subject(), self.fixture, divergence_k=10,
|
| 207 | run_root=self.tmp / "runs", run_index=0, metadata=make_metadata(),
|
| 208 | )
|
| 209 | self.assertEqual(validate_rows(rows, "contention"), [])
|
| 210 | self.assertEqual(len(rows), 1)
|
| 211 | row = rows[0]
|
| 212 | # ADR-0005: K is encoded in the immutable scenario name.
|
| 213 | self.assertEqual(row["scenario"], "contention_long_divergence_k10")
|
| 214 | self.assertEqual(row["divergence_k"], 10)
|
| 215 | self.assertEqual(row["workers"], 1)
|
| 216 | self.assertIsNotNone(row["merge_ms"])
|
| 217 | self.assertFalse(row["merge_conflict"])
|
| 218 | self.assertEqual(row["returncode"], 0)
|
| 219 | self.assertTrue(row["parallel_metrics"]["integrity_check_passed"])
|
| 220 |
|
| 221 |
|
| 222 | class PartialFlushTest(unittest.TestCase):
|
| 223 | def setUp(self) -> None:
|
| 224 | self.tmp = Path(tempfile.mkdtemp(prefix="oakbench-partial-test-"))
|
| 225 |
|
| 226 | def tearDown(self) -> None:
|
| 227 | shutil.rmtree(self.tmp, ignore_errors=True)
|
| 228 |
|
| 229 | @staticmethod
|
| 230 | def _row(run: int) -> dict:
|
| 231 | return {
|
| 232 | "bench_id": "tid",
|
| 233 | "profile": "parallel-contention",
|
| 234 | "scenario": "contention_merge_train_w2",
|
| 235 | "contention_mode": "merge_train",
|
| 236 | "workers": 2,
|
| 237 | "operation": "parallel.total",
|
| 238 | "run": run,
|
| 239 | "subject": "git",
|
| 240 | "subject_kind": "git",
|
| 241 | "elapsed_ms": 1.0,
|
| 242 | "returncode": 0,
|
| 243 | }
|
| 244 |
|
| 245 | def test_partial_exists_during_run_and_finalized_file_matches(self) -> None:
|
| 246 | store = ResultsStore(self.tmp, lane="contention", flush_partial=True)
|
| 247 | rows = [self._row(0), self._row(1)]
|
| 248 | store.append_partial("tid", rows[:1])
|
| 249 | partial = store.partial_path("tid")
|
| 250 | self.assertTrue(partial.exists(), "partial.jsonl must exist mid-run")
|
| 251 | store.append_partial("tid", rows[1:])
|
| 252 | partial_content = partial.read_text()
|
| 253 | self.assertEqual(len(partial_content.splitlines()), 2)
|
| 254 |
|
| 255 | raw_path, _ = store.write("tid", rows)
|
| 256 | self.assertEqual(raw_path.read_text(), partial_content)
|
| 257 | self.assertFalse(partial.exists(), "write() must finalize (remove) the partial")
|
| 258 | parsed = [json.loads(line) for line in raw_path.read_text().splitlines()]
|
| 259 | self.assertEqual(parsed, rows)
|
| 260 |
|
| 261 | def test_flush_partial_defaults_off_and_write_is_byte_identical(self) -> None:
|
| 262 | rows = [self._row(0)]
|
| 263 | plain = ResultsStore(self.tmp / "plain", lane="contention")
|
| 264 | with self.assertRaises(RuntimeError):
|
| 265 | plain.append_partial("tid", rows)
|
| 266 | flushing = ResultsStore(self.tmp / "flush", lane="contention", flush_partial=True)
|
| 267 | plain_path, _ = plain.write("tid", rows)
|
| 268 | flush_path, _ = flushing.write("tid", rows)
|
| 269 | # Existing-lane invariance: the canonical file is byte-identical
|
| 270 | # whether or not partial flushing is enabled.
|
| 271 | self.assertEqual(plain_path.read_bytes(), flush_path.read_bytes())
|
| 272 |
|
| 273 | def test_write_without_flush_partial_leaves_foreign_partials_alone(self) -> None:
|
| 274 | store = ResultsStore(self.tmp, lane="contention")
|
| 275 | stray = self.tmp / "tid.partial.jsonl"
|
| 276 | self.tmp.mkdir(parents=True, exist_ok=True)
|
| 277 | stray.write_text("{}\n")
|
| 278 | store.write("tid", [self._row(0)])
|
| 279 | self.assertTrue(stray.exists())
|
| 280 |
|
| 281 |
|
| 282 | @unittest.skipIf(GIT is None, "git not installed")
|
| 283 | class CliEndToEndTest(unittest.TestCase):
|
| 284 | def setUp(self) -> None:
|
| 285 | self.tmp = Path(tempfile.mkdtemp(prefix="oakbench-fleet-cli-test-"))
|
| 286 |
|
| 287 | def tearDown(self) -> None:
|
| 288 | shutil.rmtree(self.tmp, ignore_errors=True)
|
| 289 |
|
| 290 | def test_merge_train_cli_writes_canonical_rows_and_cleans_partial(self) -> None:
|
| 291 | env = os.environ.copy()
|
| 292 | env["OAK_BENCH_LOCK"] = "off"
|
| 293 | results = self.tmp / "results"
|
| 294 | proc = subprocess.run(
|
| 295 | [
|
| 296 | sys.executable,
|
| 297 | str(ROOT / "scripts" / "parallel_contention.py"),
|
| 298 | "--subjects", "git",
|
| 299 | "--workers", "2",
|
| 300 | "--commits-per-worker", "1",
|
| 301 | "--modes", "merge_train",
|
| 302 | "--results", str(results),
|
| 303 | "--workdir", str(self.tmp / "work"),
|
| 304 | ],
|
| 305 | env=env,
|
| 306 | text=True,
|
| 307 | capture_output=True,
|
| 308 | timeout=300,
|
| 309 | )
|
| 310 | self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
|
| 311 | raw_files = [p for p in results.glob("*.jsonl") if ".partial." not in p.name and p.name != "latest.jsonl"]
|
| 312 | self.assertEqual(len(raw_files), 1)
|
| 313 | self.assertEqual(list(results.glob("*.partial.jsonl")), [], "partial must be finalized away")
|
| 314 | rows = [json.loads(line) for line in raw_files[0].read_text().splitlines()]
|
| 315 | self.assertEqual(validate_rows(rows, "contention"), [])
|
| 316 | scenarios = {row["scenario"] for row in rows}
|
| 317 | self.assertEqual(scenarios, {"contention_merge_train_w2"})
|
| 318 |
|
| 319 |
|
| 320 | if __name__ == "__main__":
|
| 321 | unittest.main()
|