Fix remaining PR #3 CI flakes and base benchmark JSON

.github/workflows/benchmark.yml

@@ -129,6 +129,12 @@ jobs:
           fi
           echo "Using base benchmark config: ${CONFIG_FILE}"
           (cd "${BENCH_BASE_DIR}" && uv sync --dev)
+          # Older main lacks --json-out / json_out support on bench scripts.
+          # Overlay HEAD's tests/performance harness into the base worktree so
+          # measurements still import base ccbt (__file__ under base) but emit
+          # CI JSON artifacts the suite runner expects.
+          mkdir -p "${BENCH_BASE_DIR}/tests/performance"
+          cp -a "${{ github.workspace }}/tests/performance/." "${BENCH_BASE_DIR}/tests/performance/"
           # Use HEAD's runner script (has current CLI), but execute benchmarks in base workdir.
           uv run python dev/scripts/run_benchmark_suite.py \
             --output-dir "${BENCH_BASE_DIR}" \

dev/scripts/run_benchmark_suite.py

@@ -145,6 +145,8 @@ def _invoke(with_json_out: bool) -> subprocess.CompletedProcess[str]:
 
     completed = _invoke(with_json_out=True)
     if completed.returncode != 0:
+        # Older scripts may reject --json-out; retry without it and look for
+        # legacy artifact paths or an explicit --output-dir write.
         completed = _invoke(with_json_out=False)
 
     if completed.returncode != 0:
@@ -159,7 +161,10 @@ def _invoke(with_json_out: bool) -> subprocess.CompletedProcess[str]:
     else:
         legacy = _find_legacy_artifact(workdir, spec.benchmark_key)
         if legacy is None:
+            detail = (completed.stderr or completed.stdout or "").strip()
             msg = f"Benchmark {spec.benchmark_key} produced no JSON artifact"
+            if detail:
+                msg = f"{msg}: {detail}"
             raise RuntimeError(msg)
         payload = _normalize_payload(_load_json(legacy), spec.benchmark_key, config_name)
 

tests/integration/test_mse_tcp_server_pe_first.py

@@ -51,6 +51,7 @@ async def _run_loopback_mse_handshake(
     port: int,
     *,
     attempts: int = 2,
+    accept_mock: AsyncMock | None = None,
 ) -> None:
     """Run an outbound MSE handshake against an already-started loopback server."""
     last_error: str | None = None
@@ -71,6 +72,8 @@ async def _run_loopback_mse_handshake(
         finally:
             writer.close()
             await writer.wait_closed()
+        if accept_mock is not None and accept_mock.await_count > 0:
+            break
         if attempt + 1 < attempts:
             await asyncio.sleep(0.05)
     assert False, last_error or "MSE handshake failed"
@@ -107,7 +110,9 @@ async def _close_incoming_connection(
     try:
         await asyncio.sleep(0.05)
         port = tcp_server.sockets[0].getsockname()[1]
-        await _run_loopback_mse_handshake(info_hash, outbound_payload, port)
+        await _run_loopback_mse_handshake(
+            info_hash, outbound_payload, port, accept_mock=accept_incoming_encrypted
+        )
 
         assert accept_incoming_encrypted.await_count == 1
         accepted = accept_incoming_encrypted.await_args.args
@@ -156,7 +161,12 @@ async def _close_incoming_connection(
     try:
         await asyncio.sleep(0.05)
         port = tcp_server.sockets[0].getsockname()[1]
-        await _run_loopback_mse_handshake(target_info_hash, outbound_payload, port)
+        await _run_loopback_mse_handshake(
+            target_info_hash,
+            outbound_payload,
+            port,
+            accept_mock=accept_incoming_target,
+        )
 
         accept_incoming_ignored.assert_not_awaited()
         accept_incoming_target.assert_awaited_once()

tests/unit/resilience/test_resilience_timeout.py

@@ -25,14 +25,18 @@ def sync_function():
 
     def test_sync_function_timeout_failure(self):
         """Test synchronous function with timeout - timeout case."""
+        gate = threading.Event()
+
         @with_timeout(0.1)
         def slow_sync_function():
-            time.sleep(0.2)  # Longer than timeout
+            gate.wait(timeout=1.0)
             return "should_not_reach_here"
 
         with pytest.raises(TimeoutError, match="Operation timed out after 0.1 seconds"):
             slow_sync_function()
 
+        gate.set()
+
     def test_sync_function_timeout_exception(self):
         """Test synchronous function with timeout - exception case."""
         @with_timeout(1.0)

tests/unit/security/test_mse_handshake.py

@@ -592,29 +592,40 @@ async def responder(
         server = await asyncio.start_server(responder, "127.0.0.1", 0)
         try:
             server_port = server.sockets[0].getsockname()[1]
-            initiator_reader, initiator_writer = await asyncio.open_connection(
-                "127.0.0.1", server_port
-            )
-            try:
-                initiator = MSEHandshake(prefer_rc4=True)
-                initiator_result = await initiator.initiate_as_initiator(
-                    initiator_reader,
-                    initiator_writer,
-                    info_hash,
-                    timeout=_MSE_INTEGRATION_TIMEOUT,
-                )
-                receiver_result = await asyncio.wait_for(
-                    responder_results.get(), timeout=_MSE_INTEGRATION_TIMEOUT
+            last_error: str | None = None
+            initiator_result: MSEHandshakeResult | None = None
+            receiver_result: MSEHandshakeResult | None = None
+            for attempt in range(2):
+                initiator_reader, initiator_writer = await asyncio.open_connection(
+                    "127.0.0.1", server_port
                 )
-            finally:
-                initiator_writer.close()
-                await initiator_writer.wait_closed()
+                try:
+                    initiator = MSEHandshake(prefer_rc4=True)
+                    initiator_result = await initiator.initiate_as_initiator(
+                        initiator_reader,
+                        initiator_writer,
+                        info_hash,
+                        timeout=_MSE_INTEGRATION_TIMEOUT,
+                    )
+                    receiver_result = await asyncio.wait_for(
+                        responder_results.get(), timeout=_MSE_INTEGRATION_TIMEOUT
+                    )
+                    if initiator_result.success and receiver_result.success:
+                        break
+                    last_error = initiator_result.error or receiver_result.error
+                finally:
+                    initiator_writer.close()
+                    await initiator_writer.wait_closed()
+                if attempt == 0:
+                    await asyncio.sleep(0.05)
         finally:
             server.close()
             await server.wait_closed()
 
-        assert initiator_result.success is True
-        assert receiver_result.success is True
+        assert initiator_result is not None
+        assert receiver_result is not None
+        assert initiator_result.success is True, last_error
+        assert receiver_result.success is True, last_error
         assert initiator_result.cipher is not None
         assert receiver_result.cipher is not None
 
@@ -1266,7 +1277,7 @@ async def responder(reader: asyncio.StreamReader, writer: asyncio.StreamWriter)
             reader=reader,
             writer=writer,
             info_hash=info_hash,
-            timeout=1.0,
+            timeout=_MSE_INTEGRATION_TIMEOUT,
             initial_payload_size=0,
             info_hash_candidates=[info_hash],
         )
@@ -1277,30 +1288,41 @@ async def responder(reader: asyncio.StreamReader, writer: asyncio.StreamWriter)
     server = await asyncio.start_server(responder, "127.0.0.1", 0)
     try:
         server_port = server.sockets[0].getsockname()[1]
-        initiator_reader, initiator_writer = await asyncio.open_connection(
-            "127.0.0.1", server_port
-        )
-        try:
-            initiator_handshake = MSEHandshake()
-            initiator_result = await initiator_handshake.initiate_as_initiator(
-                initiator_reader,
-                initiator_writer,
-                info_hash,
-                timeout=1.0,
-                initial_payload=initial_payload,
-            )
-            responder_result = await asyncio.wait_for(
-                responder_results.get(), timeout=1.0
+        last_error: str | None = None
+        initiator_result: MSEHandshakeResult | None = None
+        responder_result: MSEHandshakeResult | None = None
+        for attempt in range(2):
+            initiator_reader, initiator_writer = await asyncio.open_connection(
+                "127.0.0.1", server_port
             )
+            try:
+                initiator_handshake = MSEHandshake()
+                initiator_result = await initiator_handshake.initiate_as_initiator(
+                    initiator_reader,
+                    initiator_writer,
+                    info_hash,
+                    timeout=_MSE_INTEGRATION_TIMEOUT,
+                    initial_payload=initial_payload,
+                )
+                responder_result = await asyncio.wait_for(
+                    responder_results.get(), timeout=_MSE_INTEGRATION_TIMEOUT
+                )
+                if initiator_result.success and responder_result.success:
+                    break
+                last_error = initiator_result.error or responder_result.error
+            finally:
+                initiator_writer.close()
+                await initiator_writer.wait_closed()
+            if attempt == 0:
+                await asyncio.sleep(0.05)
 
-            assert initiator_result.success is True
-            assert initiator_result.resolved_info_hash == info_hash
-            assert responder_result.success is True
-            assert responder_result.decrypted_initial_data == initial_payload
-            assert responder_result.resolved_info_hash == info_hash
-        finally:
-            initiator_writer.close()
-            await initiator_writer.wait_closed()
+        assert initiator_result is not None
+        assert responder_result is not None
+        assert initiator_result.success is True, last_error
+        assert initiator_result.resolved_info_hash == info_hash
+        assert responder_result.success is True, last_error
+        assert responder_result.decrypted_initial_data == initial_payload
+        assert responder_result.resolved_info_hash == info_hash
     finally:
         server.close()
         await server.wait_closed()

tests/unit/storage/conftest.py

@@ -0,0 +1,15 @@
+"""Shared fixtures for storage unit tests."""
+
+from __future__ import annotations
+
+import tempfile
+from pathlib import Path
+
+import pytest
+
+
+@pytest.fixture
+def temp_db_path() -> str:
+    """Provide an isolated SQLite cache path with a dedicated chunk store directory."""
+    with tempfile.TemporaryDirectory() as tmpdir:
+        yield str(Path(tmpdir) / "cache.db")

tests/unit/storage/test_xet_data_aggregator.py

@@ -5,7 +5,7 @@
 
 from __future__ import annotations
 
-import tempfile
+import hashlib
 
 import pytest
 
@@ -17,29 +17,12 @@
 
 def _chunk_hash(label: bytes) -> bytes:
     """Build a deterministic 32-byte chunk hash for tests."""
-    return (label * 16)[:32]
+    return hashlib.sha256(label).digest()
 
 
 class TestXetDataAggregator:
     """Test XetDataAggregator class."""
 
-    @pytest.fixture
-    def temp_db_path(self):
-        """Create temporary database path for testing."""
-        with tempfile.NamedTemporaryFile(delete=False, suffix=".db") as f:
-            db_path = f.name
-        yield db_path
-        # Cleanup
-        import os
-        import time
-        for _ in range(5):
-            try:
-                if os.path.exists(db_path):
-                    os.unlink(db_path)
-                break
-            except (PermissionError, OSError):
-                time.sleep(0.1)
-
     @pytest.fixture
     def dedup(self, temp_db_path):
         """Create XetDeduplication instance for testing."""

tests/unit/storage/test_xet_deduplication.py

@@ -21,22 +21,6 @@
 class TestXetDeduplication:
     """Test XetDeduplication class."""
 
-    @pytest.fixture
-    def temp_db_path(self):
-        """Create temporary database path for testing."""
-        with tempfile.NamedTemporaryFile(delete=False, suffix=".db") as f:
-            db_path = f.name
-        yield db_path
-        # Cleanup - try multiple times on Windows
-        import time
-        for _ in range(5):
-            try:
-                if os.path.exists(db_path):
-                    os.unlink(db_path)
-                break
-            except (PermissionError, OSError):
-                time.sleep(0.1)
-
     @pytest.fixture
     def dedup(self, temp_db_path):
         """Create XetDeduplication instance for testing."""
@@ -48,20 +32,22 @@ def dedup(self, temp_db_path):
     def test_initialization(self, temp_db_path):
         """Test deduplication cache initialization."""
         dedup = XetDeduplication(cache_db_path=temp_db_path)
+        try:
+            # Database should be created
+            assert os.path.exists(temp_db_path)
+
+            # Check table exists
+            conn = sqlite3.connect(temp_db_path)
+            cursor = conn.cursor()
+            cursor.execute(
+                "SELECT name FROM sqlite_master WHERE type='table' AND name='chunks'"
+            )
+            result = cursor.fetchone()
+            conn.close()
 
-        # Database should be created
-        assert os.path.exists(temp_db_path)
-
-        # Check table exists
-        conn = sqlite3.connect(temp_db_path)
-        cursor = conn.cursor()
-        cursor.execute(
-            "SELECT name FROM sqlite_master WHERE type='table' AND name='chunks'"
-        )
-        result = cursor.fetchone()
-        conn.close()
-
-        assert result is not None
+            assert result is not None
+        finally:
+            dedup.close()
 
     @pytest.mark.asyncio
     async def test_check_chunk_not_exists(self, dedup):

tests/unit/storage/test_xet_defrag_prevention.py

@@ -5,8 +5,6 @@
 
 from __future__ import annotations
 
-import tempfile
-
 import pytest
 
 from ccbt.storage.xet_deduplication import XetDeduplication
@@ -18,23 +16,6 @@
 class TestXetDefragPrevention:
     """Test XetDefragPrevention class."""
 
-    @pytest.fixture
-    def temp_db_path(self):
-        """Create temporary database path for testing."""
-        with tempfile.NamedTemporaryFile(delete=False, suffix=".db") as f:
-            db_path = f.name
-        yield db_path
-        # Cleanup
-        import os
-        import time
-        for _ in range(5):
-            try:
-                if os.path.exists(db_path):
-                    os.unlink(db_path)
-                break
-            except (PermissionError, OSError):
-                time.sleep(0.1)
-
     @pytest.fixture
     def dedup(self, temp_db_path):
         """Create XetDeduplication instance for testing."""