From aec0b4670f33b25b26652df9d4cd7a574a4b60bd Mon Sep 17 00:00:00 2001 From: Codex Date: Mon, 13 Jul 2026 08:11:54 +0200 Subject: [PATCH] =?UTF-8?q?dita=5Fv2(test):=20repair=20test=5Fdouble=5Fbuf?= =?UTF-8?q?fer=20control=20test=20=E2=80=94=20suite=20was=20un-collectable?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Line 154 held mutant debris: KernelControlSnapshot(mode="live", mode="live"") — duplicate kwarg + unterminated string. Committed since a55501b, it made the ENTIRE dita_v2 suite fail collection, so nobody has run it. That is why no test caught the bingx_venue telemetry signature skew that orphaned 6 live positions tonight. The test was fantasy besides: asserted read.arming == "DARK" (no such field) and mode == "live" (KernelMode is NORMAL|DEBUG). Rewritten to assert a real invariant — the second write must flip the buffer and become visible. Co-Authored-By: Claude Opus 4.8 --- prod/clean_arch/dita_v2/test_double_buffer.py | 231 ++++++++++++++++++ 1 file changed, 231 insertions(+) create mode 100644 prod/clean_arch/dita_v2/test_double_buffer.py diff --git a/prod/clean_arch/dita_v2/test_double_buffer.py b/prod/clean_arch/dita_v2/test_double_buffer.py new file mode 100644 index 0000000..005da89 --- /dev/null +++ b/prod/clean_arch/dita_v2/test_double_buffer.py @@ -0,0 +1,231 @@ +"""Phase-4 tests: double-buffer region — torn-read impossible under concurrent publish. + +Key invariants: +1. read() never returns a partially-updated value +2. After write(), subsequent reads return the written value +3. Concurrent hammering: reads under heavy write load never get torn data +4. Deep-copy isolation: read returns independent copy +""" + +from __future__ import annotations + +import sys +import threading +import time + +sys.path.insert(0, "/mnt/dolphinng5_predict") + +from prod.clean_arch.dita_v2.contracts import AccountStateSnapshot, VenueTelemetrySnapshot +from prod.clean_arch.dita_v2.double_buffer import DoubleBufferRegion +from prod.clean_arch.dita_v2.zinc_plane import InMemoryZincPlane + + +class TestDoubleBufferBasic: + """Basic write/read cycle with DoubleBufferRegion.""" + + def test_write_then_read(self) -> None: + buf = DoubleBufferRegion(lambda: AccountStateSnapshot()) + snap = AccountStateSnapshot(wallet_balance=25000.0, e_live=True) + buf.write(snap) + read = buf.read() + assert read.wallet_balance == 25000.0 + assert read.e_live is True + + def test_overwrite(self) -> None: + buf = DoubleBufferRegion(lambda: AccountStateSnapshot()) + buf.write(AccountStateSnapshot(wallet_balance=100.0, e_live=True)) + assert buf.read().wallet_balance == 100.0 + buf.write(AccountStateSnapshot(wallet_balance=200.0, e_live=True)) + assert buf.read().wallet_balance == 200.0 + + def test_initial_read_is_default(self) -> None: + buf = DoubleBufferRegion(lambda: AccountStateSnapshot(e_live=False)) + assert buf.read().e_live is False + assert buf.read().wallet_balance == 0.0 + + def test_read_returns_deep_copy(self) -> None: + """read() returns a copy — modifying it doesn't corrupt the buffer.""" + buf = DoubleBufferRegion(lambda: AccountStateSnapshot(wallet_balance=100.0)) + r1 = buf.read() + # Attempt to mutate the returned object (frozen, so conceptually) + assert r1.wallet_balance == 100.0 + buf.write(AccountStateSnapshot(wallet_balance=200.0)) + # Previous read result unaffected + assert r1.wallet_balance == 100.0 + + +class TestDoubleBufferTornRead: + """Under heavy concurrent publish/read, no torn reads. + + Mutation: remove double-buffer (use single slot) → reads may see partial + state under write load. + """ + + def test_no_torn_read_under_writer(self) -> None: + """1000 rapid writes interleaved with 1000 reads — all reads return + a consistent snapshot that was actually published.""" + plane = InMemoryZincPlane() + seen: list[AccountStateSnapshot] = [] + errors: list[str] = [] + lock = threading.Lock() + + def writer() -> None: + for i in range(100): + plane.publish_account(AccountStateSnapshot( + wallet_balance=float(i * 100), + available_margin=float(i * 80), + used_margin=float(i * 20), + event_seq=i, mono_ns=i, + e_live=True, reconcile_ok=(i % 2 == 0), + )) + + def reader() -> None: + for _ in range(100): + try: + snap = plane.read_account() + # A torn read would show inconsistent fields. + # wallet_balance should always be a multiple of 100 + # between 0 and 9900 + wb = snap.wallet_balance + if isinstance(wb, float): + pass + except Exception as e: + with lock: + errors.append(str(e)) + + threads = [threading.Thread(target=writer, daemon=True) for _ in range(5)] + threads += [threading.Thread(target=reader, daemon=True) for _ in range(10)] + for t in threads: + t.start() + for t in threads: + t.join(timeout=10) + assert len(errors) == 0, f"errors: {errors}" + + def test_no_torn_read_rapid_alternating(self) -> None: + """Interleaved write/read at high frequency — never returns garbage.""" + plane = InMemoryZincPlane() + for i in range(5000): + plane.publish_account(AccountStateSnapshot( + wallet_balance=float(i * 1000), + available_margin=float(i * 800), + used_margin=float(i * 200), + event_seq=i, mono_ns=i, + e_live=True, reconcile_ok=True, + )) + snap = plane.read_account() + # Must be internally consistent: all fields match same event_seq + assert snap.event_seq == i, f"seq mismatch: {snap.event_seq} != {i}" + assert snap.wallet_balance == float(i * 1000), f"wb mismatch" + assert plane.read_account().event_seq == 4999 + + def test_concurrent_no_torn_on_venue(self) -> None: + """Venue region double-buffer also protects against torn reads.""" + plane = InMemoryZincPlane() + errors: list[str] = [] + lock = threading.Lock() + + def vwriter() -> None: + for i in range(100): + plane.publish_venue(VenueTelemetrySnapshot( + phase="live", status=f"S{i}", venue="bingx", + )) + + def vreader() -> None: + for _ in range(100): + try: + snap = plane.read_venue() + _ = snap.phase + except Exception as e: + with lock: + errors.append(str(e)) + + threads = [threading.Thread(target=vwriter, daemon=True) for _ in range(5)] + threads += [threading.Thread(target=vreader, daemon=True) for _ in range(5)] + for t in threads: + t.start() + for t in threads: + t.join(timeout=10) + assert len(errors) == 0 + + def test_control_region_double_buffered(self) -> None: + """Control region round-trips, and a second write flips the buffer.""" + from prod.clean_arch.dita_v2.control import KernelControlSnapshot, KernelMode + + plane = InMemoryZincPlane() + plane.update_control(KernelControlSnapshot(mode=KernelMode.DEBUG)) + assert plane.read_control().mode is KernelMode.DEBUG + # The buffer must FLIP: a later write has to become visible, not be + # shadowed by the first snapshot. + plane.update_control(KernelControlSnapshot(mode=KernelMode.NORMAL)) + assert plane.read_control().mode is KernelMode.NORMAL + + def test_two_regions_independent(self) -> None: + """Account and venue regions don't interfere.""" + plane = InMemoryZincPlane() + plane.publish_account(AccountStateSnapshot(wallet_balance=100.0, e_live=True)) + plane.publish_venue(VenueTelemetrySnapshot(phase="test")) + assert plane.read_account().wallet_balance == 100.0 + assert plane.read_venue().phase == "test" + + def test_deep_copy_independence(self) -> None: + """Multiple reads from the same buffer return independent snapshots.""" + from prod.clean_arch.dita_v2.contracts import VenueTelemetrySnapshot + buf = DoubleBufferRegion(lambda: VenueTelemetrySnapshot()) + buf.write(VenueTelemetrySnapshot(phase="original")) + r1 = buf.read() + r2 = buf.read() + assert r1.phase == "original" + assert r2.phase == "original" + # They should be equal but independent + assert r1 == r2 + + +class TestDoubleBufferConcurrent: + """Heavy concurrent stress on DoubleBufferRegion.""" + + def test_50_writers_50_readers(self) -> None: + """50 threads writing, 50 threads reading — no crash.""" + from prod.clean_arch.dita_v2.contracts import VenueTelemetrySnapshot + buf = DoubleBufferRegion(lambda: VenueTelemetrySnapshot()) + errors: list[Exception] = [] + lock = threading.Lock() + + def writer() -> None: + try: + for i in range(50): + buf.write(VenueTelemetrySnapshot(phase=f"run_{i}")) + except Exception as e: + with lock: + errors.append(e) + + def reader() -> None: + try: + for _ in range(50): + _ = buf.read() + except Exception as e: + with lock: + errors.append(e) + + threads = [threading.Thread(target=writer, daemon=True) for _ in range(50)] + threads += [threading.Thread(target=reader, daemon=True) for _ in range(50)] + for t in threads: + t.start() + for t in threads: + t.join(timeout=15) + assert len(errors) == 0, f"{len(errors)} errors" + + def test_rapid_alternating_10k(self) -> None: + """10k alternating write/read on a fresh buffer — always returns a valid value.""" + buf = DoubleBufferRegion(lambda: AccountStateSnapshot()) + for i in range(5000): + buf.write(AccountStateSnapshot( + wallet_balance=float(i * 100), + available_margin=float(i * 80), + used_margin=float(i * 20), + event_seq=i, mono_ns=i, + e_live=True, reconcile_ok=(i % 2 == 0), + )) + snap = buf.read() + assert snap.event_seq == i + assert snap.wallet_balance == float(i * 100) + assert buf.read().event_seq == 4999