diff --git a/pycbsdk/pyproject.toml b/pycbsdk/pyproject.toml index b57a8bfd..4f715fcc 100644 --- a/pycbsdk/pyproject.toml +++ b/pycbsdk/pyproject.toml @@ -49,6 +49,8 @@ lint = [ test = [ "pytest", "pytest-timeout", + "numpy", + "pylsl", ] [tool.setuptools.packages.find] @@ -60,6 +62,7 @@ pycbsdk = ["*.dll", "*.so", "*.so.*", "*.dylib"] [tool.pytest.ini_options] markers = [ "integration: tests requiring a running nPlayServer instance", + "lsl: nPlay+LSL tests; need an EXCLUSIVE nPlayServer (one shmem per device type). Auto-skipped when file-based nPlay tests are also collected — run with `pytest -m lsl`.", ] [tool.setuptools_scm] diff --git a/pycbsdk/src/pycbsdk/_cdef.py b/pycbsdk/src/pycbsdk/_cdef.py index ae7be8c1..fdc44cbd 100644 --- a/pycbsdk/src/pycbsdk/_cdef.py +++ b/pycbsdk/src/pycbsdk/_cdef.py @@ -357,6 +357,10 @@ cbsdk_result_t cbsdk_session_get_clock_offset(cbsdk_session_t session, int64_t* offset_ns); cbsdk_result_t cbsdk_session_get_clock_uncertainty(cbsdk_session_t session, int64_t* uncertainty_ns); cbsdk_result_t cbsdk_session_send_clock_probe(cbsdk_session_t session); +cbsdk_result_t cbsdk_session_to_local_time( + cbsdk_session_t session, int64_t stream_id, + const uint64_t* device_ns, int64_t* out_steady_ns, size_t n); +cbsdk_result_t cbsdk_session_reset_monotonic(cbsdk_session_t session, int64_t stream_id); // Utility int64_t cbsdk_get_steady_clock_ns(void); diff --git a/pycbsdk/src/pycbsdk/session.py b/pycbsdk/src/pycbsdk/session.py index 04037785..ece0d22b 100644 --- a/pycbsdk/src/pycbsdk/session.py +++ b/pycbsdk/src/pycbsdk/session.py @@ -1712,15 +1712,62 @@ def send_clock_probe(self): "Failed to send clock probe", ) - def device_to_monotonic(self, device_time_ns: int) -> float: + def device_to_monotonic_batch(self, device_ns, stream_id: int = -1) -> list[float]: + """Convert a batch of device timestamps to ``time.monotonic()`` seconds. + + The whole batch is converted inside the library against one consistent + clock snapshot (device_ns → steady_clock_ns), then the steady_clock → + monotonic offset is applied here. This is the efficient path for + per-batch stamping and the only way to get monotonicity enforcement. + + Args: + device_ns: Iterable of device timestamps in nanoseconds. + stream_id: Per-stream key controlling monotonicity. When ``>= 0``, + each output is clamped to be non-decreasing relative to the + previous conversion request on the *same* ``stream_id`` — except + across a genuine clock re-sync, where the floor resets so the + timeline follows the new regime instead of stalling. The + natural key is the sample group (e.g. one id for the 30 kHz raw + stream, another for 1 kHz); use a distinct id per derived stream. + ``-1`` (default) is a stateless conversion with no clamping — + use it for spikes and one-off lookups. Monotonic streams assume + timestamps are submitted in intended-output order. + + Returns: + List of ``time.monotonic()`` values in seconds, one per input. + + Raises: + RuntimeError: If no clock sync data is available yet. + """ + self._maybe_recalibrate_monotonic() + device_list = [int(t) for t in device_ns] + n = len(device_list) + if n == 0: + return [] + inp = ffi.new("uint64_t[]", device_list) + out = ffi.new("int64_t[]", n) + result = _get_lib().cbsdk_session_to_local_time( + self._session, stream_id, inp, out, n + ) + if result != 0: + raise RuntimeError("No clock sync data available") + k = self._mono_to_steady_offset_ns + return [(out[i] - k) / 1_000_000_000 for i in range(n)] + + def device_to_monotonic(self, device_time_ns: int, stream_id: int = -1) -> float: """Convert a device timestamp to ``time.monotonic()`` seconds. Chains two offsets: - 1. device_ns → steady_clock_ns (via clock_offset_ns from device sync) + 1. device_ns → steady_clock_ns (via the device clock sync) 2. steady_clock_ns → monotonic_ns (via calibration at session creation) Args: device_time_ns: Device timestamp in nanoseconds (e.g., header.time). + stream_id: Per-stream monotonicity key; ``-1`` (default) is a + stateless conversion. See :meth:`device_to_monotonic_batch` for + the monotonicity semantics; for a stream you want kept monotonic, + stamp each batch through that method (or pass a stable + ``stream_id`` here) rather than mixing ids. Returns: Corresponding ``time.monotonic()`` value in seconds. @@ -1736,13 +1783,18 @@ def on_spike(header, data): latency_ms = (time.monotonic() - t) * 1000 print(f"Spike latency: {latency_ms:.1f} ms") """ - self._maybe_recalibrate_monotonic() - offset = self.clock_offset_ns - if offset is None: - raise RuntimeError("No clock sync data available") - steady_ns = device_time_ns - offset - mono_ns = steady_ns - self._mono_to_steady_offset_ns - return mono_ns / 1_000_000_000 + return self.device_to_monotonic_batch([device_time_ns], stream_id)[0] + + def reset_monotonic(self, stream_id: int) -> None: + """Drop the monotonic floor/epoch state for one ``stream_id``. + + The next monotonic conversion on that stream starts fresh. No-op if the + stream is unknown. + """ + _check( + _get_lib().cbsdk_session_reset_monotonic(self._session, stream_id), + "Failed to reset monotonic state", + ) # --- Commands --- diff --git a/pycbsdk/tests/_lsl_generator.py b/pycbsdk/tests/_lsl_generator.py new file mode 100644 index 00000000..bac8974b --- /dev/null +++ b/pycbsdk/tests/_lsl_generator.py @@ -0,0 +1,72 @@ +"""Standalone 30 kHz int16 LSL stream generator for nPlay+LSL tests. + +Launched as a subprocess by the ``nplay_lsl`` fixture. nPlayServer +(``--device=LSL ``) resolves this stream by name and replays it onto the +Cerebus protocol, deriving each packet's device timestamp from the stream's +ORIGINATING time scaled to a 30 kHz sample counter. We therefore pace pushes to +real time so the originating timestamps advance at the true sample rate — the +condition CereLink's clock sync must map to a monotonic host timeline. + +Usage: + python _lsl_generator.py [n_channels] + +Runs until killed (SIGTERM/SIGINT) by the fixture. +""" + +import signal +import sys +import time + +import numpy as np +import pylsl + +SRATE = 30000 +CHUNK_SAMPLES = 300 # 10 ms per push at 30 kHz + + +def main() -> int: + name = sys.argv[1] + n_ch = int(sys.argv[2]) if len(sys.argv) > 2 else 8 + + info = pylsl.StreamInfo( + name=name, + type="EEG", + channel_count=n_ch, + nominal_srate=SRATE, + channel_format=pylsl.cf_int16, + source_id=f"{name}_src", + ) + outlet = pylsl.StreamOutlet(info, chunk_size=CHUNK_SAMPLES, max_buffered=360) + + # Exit cleanly on the fixture's terminate() so liblsl tears the outlet down. + running = {"go": True} + + def _stop(*_a): + running["go"] = False + + signal.signal(signal.SIGTERM, _stop) + signal.signal(signal.SIGINT, _stop) + + # A slow sine across channels — content is irrelevant to the timing test, + # but non-constant data makes the stream easy to eyeball in a viewer. + phase = np.arange(CHUNK_SAMPLES, dtype=np.float64) + chan_offsets = np.linspace(0.0, np.pi, n_ch, dtype=np.float64) + + t0 = time.monotonic() + idx = 0 + while running["go"]: + angle = 2.0 * np.pi * (phase[:, None] + idx) / 3000.0 + chan_offsets[None, :] + chunk = (np.sin(angle) * 1000.0).astype(np.int16) + outlet.push_chunk(chunk) + idx += CHUNK_SAMPLES + # Pace to real time: sample idx should be emitted at t0 + idx/SRATE. + target = t0 + idx / SRATE + dt = target - time.monotonic() + if dt > 0: + time.sleep(dt) + + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/pycbsdk/tests/conftest.py b/pycbsdk/tests/conftest.py index 5f03ba37..740e56d0 100644 --- a/pycbsdk/tests/conftest.py +++ b/pycbsdk/tests/conftest.py @@ -17,7 +17,7 @@ import pytest -CERELINK_RELEASE_URL = "https://github.com/CerebusOSS/CereLink/releases/download/v9.3.0" +CERELINK_RELEASE_URL = "https://github.com/CerebusOSS/CereLink/releases/download/v9.11.0" CACHE_DIR = Path(__file__).parent / ".test_cache" @@ -41,6 +41,89 @@ def _sem_unlink(name: str) -> None: libc.sem_unlink(sem_name) # ignore errors (ENOENT is fine) +def _remove_lockfile(name: str) -> None: + """Remove nPlayServer's lock FILE ($TMPDIR/.lock). + + nPlayServer's lock is a file guarded by O_EXCL (not a POSIX semaphore), so a + hard-killed instance leaves it behind — which makes the next launch with the + same name fail with "Cannot run multiple instances". Safe when absent. + """ + if platform.system() == "Windows": + return + tmp = os.environ.get("TMPDIR") or "/tmp" + (Path(tmp.rstrip("/")) / f"{name}.lock").unlink(missing_ok=True) + + +def pytest_configure(config): + """Fail fast if the DEV-built libcbsdk is OLDER than the C++ sources. + + In a source checkout pycbsdk loads ``cmake-build-*/src/cbsdk/libcbsdk.dylib`` + (there is no wheel-bundled lib). If that dylib was not rebuilt after a C++ + change, the tests silently run STALE code — which once masked a clock-sync + fix for an entire debugging session. Only a build-tree dylib is checked (an + installed/wheel-bundled lib has no sources to compare against); bypass with + ``PYCBSDK_ALLOW_STALE_LIB=1``. + """ + if os.environ.get("PYCBSDK_ALLOW_STALE_LIB"): + return + try: + from pycbsdk._lib import _find_library + lib_path = Path(_find_library()) + except Exception: + return # can't resolve the lib — let the normal load path report it + lib_str = str(lib_path) + if "cmake-build" not in lib_str and f"{os.sep}build{os.sep}" not in lib_str: + return # installed/bundled lib, not a dev build tree + if not lib_path.exists(): + return + repo_root = Path(__file__).resolve().parents[2] + src_root = repo_root / "src" + if not src_root.exists(): + return + lib_mtime = lib_path.stat().st_mtime + newest, newest_file = lib_mtime, None + for f in src_root.rglob("*"): + if f.suffix in (".cpp", ".h", ".hpp", ".c", ".cc") and f.is_file(): + m = f.stat().st_mtime + if m > newest: + newest, newest_file = m, f + if newest_file is not None: + raise pytest.UsageError( + f"Stale libcbsdk: {lib_path}\n" + f" built {time.ctime(lib_mtime)} is OLDER than C++ source " + f"{newest_file.relative_to(repo_root)} ({time.ctime(newest)}).\n" + f" Tests would run pre-change code. Rebuild the dev dylib:\n" + f" cmake --build {lib_path.parents[2]} --target cbsdk_shared\n" + f" (or set PYCBSDK_ALLOW_STALE_LIB=1 to bypass this check)." + ) + + +@pytest.hookimpl(trylast=True) +def pytest_collection_modifyitems(config, items): + """Auto-skip ``lsl`` tests when file-based nPlay tests are also collected. + + Both bring up a STANDALONE ``Session(DeviceType.NPLAY)`` whose shared memory + is keyed by device type, so two nPlayServer instances cannot coexist. When a + single run collects both, skip the LSL ones with guidance to run them alone. + + Runs ``trylast`` so pytest's own ``-m``/``-k`` deselection happens FIRST: + under ``pytest -m lsl`` the file-based tests are already removed from + ``items``, so the LSL tests are NOT skipped and actually run. + """ + lsl_items = [it for it in items if it.get_closest_marker("lsl")] + nplay_file_items = [ + it for it in items + if it.get_closest_marker("integration") and not it.get_closest_marker("lsl") + ] + if lsl_items and nplay_file_items: + skip = pytest.mark.skip( + reason="LSL nPlay tests need an exclusive nPlayServer " + "(one shmem per device type); run separately: pytest -m lsl" + ) + for it in lsl_items: + it.add_marker(skip) + + def _nplay_asset_name() -> str | None: """Return the nPlayServer zip asset name for this platform, or None.""" system = platform.system() @@ -162,11 +245,29 @@ def nplayserver_binary() -> Path | None: for candidate in extract_dir.rglob(f"nPlayServer{exe_suffix}"): if candidate.is_file(): if platform.system() != "Windows": - candidate.chmod(0o755) + _ensure_executable(candidate) return candidate return None +def _ensure_executable(path: Path) -> None: + """Make *path* executable, tolerating a transient EPERM on macOS. + + A freshly-downloaded SIGNED binary is briefly locked by Gatekeeper / + syspolicyd on first access, so chmod can raise PermissionError. The file is + usually already 0755 (preserved by the archive), so we skip when it's already + executable and otherwise retry a few times before giving up. + """ + for _ in range(5): + if os.access(path, os.X_OK): + return + try: + path.chmod(0o755) + return + except PermissionError: + time.sleep(0.5) + + @pytest.fixture(scope="session") def nplayserver(nplayserver_binary: Path | None, ns6_path: Path): """Start nPlayServer playing back test data; kill on teardown. @@ -244,3 +345,105 @@ def client_session(nplay_session): "is the nplay_session fixture alive?" ) yield session + + +# --------------------------------------------------------------------------- +# nPlay + LSL fixtures (mutually exclusive with the file-playback fixtures above; +# see pytest_collection_modifyitems). Run with: pytest -m lsl +# --------------------------------------------------------------------------- + +@pytest.fixture(scope="session") +def lsl_stream_name() -> str: + """Unique LSL stream name for this test process.""" + return f"pycbsdk_lsl_{os.getpid()}" + + +@pytest.fixture(scope="session") +def lsl_generator(lsl_stream_name: str): + """Launch the 30 kHz int16 LSL generator and wait until its stream resolves. + + Skips dependent tests if pylsl is unavailable. Yields the stream name. + """ + pylsl = pytest.importorskip("pylsl", reason="pylsl required for LSL tests") + + gen_script = Path(__file__).parent / "_lsl_generator.py" + proc = subprocess.Popen( + [sys.executable, str(gen_script), lsl_stream_name, "8"], + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + ) + + streams = pylsl.resolve_byprop("name", lsl_stream_name, timeout=10) + if not streams: + proc.terminate() + proc.wait() + out = proc.stdout.read().decode(errors="replace") if proc.stdout else "" + pytest.fail(f"LSL generator stream '{lsl_stream_name}' never appeared\n{out}") + + yield lsl_stream_name + + proc.terminate() + try: + proc.wait(timeout=5) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait() + + +@pytest.fixture(scope="session") +def nplay_lsl(nplayserver_binary: Path | None, lsl_generator: str): + """Start nPlayServer as an LSL inlet reading the generator stream. + + nPlayServer derives device timestamps from the stream's originating time + (× 30 kHz), so this exercises the LSL clock path rather than file playback. + """ + if nplayserver_binary is None: + pytest.skip( + f"nPlayServer not available for {platform.system()}/{platform.machine()}" + ) + + lock_name = f"nplay_lsl_{os.getpid()}" + _remove_lockfile(lock_name) # clear any stale lock from a hard-killed prior run + + proc = subprocess.Popen( + [ + str(nplayserver_binary), + "--audio", "none", + "--lockfile", lock_name, + "--device=LSL", lsl_generator, + "-A", # autostart + ], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + + # Allow time to resolve the LSL inlet and bind its UDP ports. + time.sleep(3) + + if proc.poll() is not None: + stdout = proc.stdout.read().decode(errors="replace") if proc.stdout else "" + stderr = proc.stderr.read().decode(errors="replace") if proc.stderr else "" + _remove_lockfile(lock_name) + pytest.fail( + f"nPlayServer --device=LSL exited immediately (rc={proc.returncode})\n" + f"stdout: {stdout}\nstderr: {stderr}" + ) + + yield proc + + proc.terminate() + try: + proc.wait(timeout=5) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait() + _remove_lockfile(lock_name) + + +@pytest.fixture(scope="session") +def nplay_lsl_session(nplay_lsl): + """A STANDALONE Session connected to the LSL-fed nPlayServer.""" + from pycbsdk import DeviceType, Session + + with Session(DeviceType.NPLAY) as session: + yield session diff --git a/pycbsdk/tests/test_lsl_sync.py b/pycbsdk/tests/test_lsl_sync.py new file mode 100644 index 00000000..4e115f3f --- /dev/null +++ b/pycbsdk/tests/test_lsl_sync.py @@ -0,0 +1,288 @@ +"""nPlay + LSL clock-synchronization stress test. + +nPlayServer (``--device=LSL ``) replays a live LSL stream onto the Cerebus +protocol, deriving each packet's device timestamp from the stream's ORIGINATING +time scaled to a 30 kHz sample counter. Unlike file playback (a sample counter +that RESETS every time the short file wraps), the LSL device clock increases +monotonically for the whole run — so CereLink's host-mapped timeline must be +monotonic across the ENTIRE capture, not just within wrap segments. + +This is the path where the ±0.6 s backward-jump bug was reported (the cursor +pipeline, 2026-06): the offset estimate recalibrated against a replay clock that +did not advance at true wall-rate, so ``device_to_monotonic`` stepped backward +and stalled any downstream consumer that assumes monotonic time. Here we stamp +every 30 kHz batch with ``device_to_monotonic(first_ts)`` — exactly as a consumer +does — and require the resulting timeline to be monotonic. + +MUTUALLY EXCLUSIVE with the file-playback tests (both use a STANDALONE NPLAY +session, whose shmem is keyed by device type). Run in isolation:: + + pytest -m lsl +""" + +import os +import time + +import pytest + +from pycbsdk import ChannelType, DeviceType, SampleRate, Session + +pytestmark = [pytest.mark.integration, pytest.mark.lsl] + +N_CHANS = 8 +WARMUP_S = 3 # let channels configure and the first offset settle +CAPTURE_S = 15 # long enough to catch a recalibration thrash + +# A backward jump in DEVICE time larger than this would mark a clock reset. The +# LSL origin clock does not wrap, so we expect ZERO of these — asserting it makes +# an unexpected wrap visible instead of silently splitting the timeline. +WRAP_BACK_S = 0.050 + +# After any segment start, ignore this much device time so a settling offset is +# not mistaken for a bug. (For LSL there is one segment; this just guards the +# very first samples after warmup.) +GUARD_S = 0.150 + +# Max tolerated BACKWARD step in the host-mapped timeline. Healthy localhost +# sync tracks the origin clock ~1:1 and never steps back; the guarded-against bug +# jumps ~600 ms, so 20 ms cleanly separates them while tolerating sub-deadband +# offset corrections. +MONO_BACKSTEP_S = 0.020 + + +def _analyze(dev_ns, mono_s, backstep_threshold_s=MONO_BACKSTEP_S): + """Walk a captured (device_ns, host_monotonic_s) timeline. + + Splits on device-clock resets (none expected for LSL), skips a short guard at + each segment start, and counts BACKWARD steps in the host-mapped time beyond + ``backstep_threshold_s``. The stateless mapping uses the tolerant default; + the per-stream monotonic mapping is checked with 0 (it must never step back). + Returns a stats dict. + """ + n = len(dev_ns) + stats = {"n_samples": 0, "n_wraps": 0, "n_segments": 0, + "n_backsteps": 0, "max_backstep_s": 0.0} + + # Segment boundaries from the device-time series. + bounds = [0] + for i in range(1, n): + if dev_ns[i] - dev_ns[i - 1] < -WRAP_BACK_S * 1e9: + stats["n_wraps"] += 1 + bounds.append(i) + bounds.append(n) + + for s in range(len(bounds) - 1): + begin, end = bounds[s], bounds[s + 1] + if end <= begin: + continue + seg_dev0 = dev_ns[begin] + prev = None + evaluated = 0 + first_eval = last_eval = None + for i in range(begin, end): + # Skip the post-(re)acquisition guard at the segment start. + if (dev_ns[i] - seg_dev0) < GUARD_S * 1e9: + prev = None + continue + if first_eval is None: + first_eval = dev_ns[i] + last_eval = dev_ns[i] + evaluated += 1 + if prev is not None: + backstep = prev - mono_s[i] # >0 => went backward + if backstep > stats["max_backstep_s"]: + stats["max_backstep_s"] = backstep + if backstep > backstep_threshold_s: + stats["n_backsteps"] += 1 + prev = mono_s[i] + + # Only count segments with a meaningful post-guard span. + if first_eval is not None and (last_eval - first_eval) > 0.3e9: + stats["n_segments"] += 1 + stats["n_samples"] += evaluated + + return stats + + +def _start_capture(session, stream_id=-1): + """Register a 30 kHz batch capture on *session*. + + Stamps each batch's first sample onto the host monotonic clock — the way an + acquisition consumer does — and returns the (device_ns, host_monotonic_s) + lists it appends to. ``stream_id < 0`` (default) uses the stateless + ``device_to_monotonic``; ``stream_id >= 0`` uses the monotonicity-enforcing + ``device_to_monotonic_batch`` keyed on that stream. + """ + dev_ns: list[int] = [] + mono_s: list[float] = [] + + @session.on_group_batch(SampleRate.SR_30kHz) + def _on_batch(samples, timestamps): + if len(timestamps) == 0: + return + t0 = int(timestamps[0]) + try: + if stream_id < 0: + m = session.device_to_monotonic(t0) + else: + m = session.device_to_monotonic_batch([t0], stream_id)[0] + except RuntimeError: + return # no clock offset yet — the safe fallback + dev_ns.append(t0) + mono_s.append(m) + + return dev_ns, mono_s + + +def _hardware_multidevice_enabled() -> bool: + """True if the operator opted in to driving a live Hub (CERELINK_HW_MULTIDEVICE).""" + return bool(os.environ.get("CERELINK_HW_MULTIDEVICE")) + + +def test_lsl_host_mapped_timeline_monotonic(nplay_lsl_session): + """device_to_monotonic(first_ts) must be monotonic across the LSL run.""" + session = nplay_lsl_session + + session.set_sample_group( + N_CHANS, ChannelType.FRONTEND, SampleRate.SR_30kHz, disable_others=True, + ) + time.sleep(WARMUP_S) + + dev_ns, mono_s = _start_capture(session) + time.sleep(CAPTURE_S) + + # Snapshot (the callback keeps firing on the shared session). + dev = list(dev_ns) + mono = list(mono_s) + assert len(dev) > 1000, f"Too few batches captured ({len(dev)}) — no data flowing?" + + stats = _analyze(dev, mono) + print( + f"\n=== nPlay+LSL timeline: {len(dev)} batches, {stats['n_wraps']} wraps, " + f"{stats['n_segments']} segments, {stats['n_backsteps']} backstep(s) " + f"> {MONO_BACKSTEP_S * 1e3:.0f} ms, max backstep = " + f"{stats['max_backstep_s'] * 1e3:.3f} ms ===" + ) + + # The LSL origin clock is monotonic and never resets. + assert stats["n_wraps"] == 0, ( + f"Unexpected device-clock reset(s) ({stats['n_wraps']}) on the LSL path — " + "the origin timestamp should increase monotonically" + ) + assert stats["n_segments"] > 0, "No evaluable timeline captured" + + assert stats["n_backsteps"] == 0, ( + f"{stats['n_backsteps']} backward jump(s) in nPlay's host-mapped LSL " + f"timeline (max {stats['max_backstep_s'] * 1e3:.3f} ms). The offset is " + "recalibrating against the replayed origin clock — the non-monotonic " + "timestamp bug that stalls downstream RESAMPLE/MERGE." + ) + + +def test_lsl_monotonic_batch_strictly_nondecreasing(nplay_lsl_session): + """device_to_monotonic_batch(stream_id) must be EXACTLY non-decreasing. + + The stateless mapping is only monotonic to within sub-deadband offset wiggle; + the per-stream monotonic API clamps every backward step, so the host timeline + must never decrease at all (checked with threshold 0). + """ + session = nplay_lsl_session + + session.set_sample_group( + N_CHANS, ChannelType.FRONTEND, SampleRate.SR_30kHz, disable_others=True, + ) + time.sleep(WARMUP_S) + + stream_id = 1 # the 30 kHz raw stream + dev_ns, mono_s = _start_capture(session, stream_id=stream_id) + time.sleep(CAPTURE_S) + + dev = list(dev_ns) + mono = list(mono_s) + assert len(dev) > 1000, f"Too few batches captured ({len(dev)}) — no data flowing?" + + stats = _analyze(dev, mono, backstep_threshold_s=0.0) + print( + f"\n=== nPlay+LSL MONOTONIC timeline: {len(dev)} batches, " + f"{stats['n_wraps']} wraps, {stats['n_segments']} segments, " + f"{stats['n_backsteps']} backstep(s), max backstep = " + f"{stats['max_backstep_s'] * 1e3:.6f} ms ===" + ) + + assert stats["n_wraps"] == 0, ( + f"Unexpected device-clock reset(s) ({stats['n_wraps']}) on the LSL path" + ) + assert stats["n_segments"] > 0, "No evaluable timeline captured" + + assert stats["n_backsteps"] == 0, ( + f"{stats['n_backsteps']} backward step(s) in the MONOTONIC host timeline " + f"(max {stats['max_backstep_s'] * 1e3:.6f} ms). " + "device_to_monotonic_batch(stream_id) must clamp every backstep." + ) + + +def test_lsl_nplay_and_hub2_timelines_monotonic(nplay_lsl_session): + """LSL-fed nPlay + a LIVE Hub2: both device clocks map to one monotonic host clock. + + The cross-device merge topology with nPlay driven from LSL instead of a file. + nPlay's device clock is the LSL stream's originating time (× 30 kHz) while + Hub2's is PTP — two unrelated time bases that each ClockSync maps onto the + SAME host monotonic clock. If nPlay's LSL-derived timeline goes non-monotonic + while Hub2's real clock stays clean, a consumer resampling one onto the other + stalls. Hardware-gated: bringing up Hub2 runs a handshake that may HARDRESET + it, so this is opt-in. + """ + if not _hardware_multidevice_enabled(): + pytest.skip( + "Set CERELINK_HW_MULTIDEVICE=1 to run against a live Hub2 alongside " + "the LSL-fed nPlay (the Hub handshake may HARDRESET it)." + ) + + nplay = nplay_lsl_session + nplay.set_sample_group( + N_CHANS, ChannelType.FRONTEND, SampleRate.SR_30kHz, disable_others=True, + ) + + with Session(DeviceType.HUB2) as hub2: + hub2.set_sample_group( + N_CHANS, ChannelType.FRONTEND, SampleRate.SR_30kHz, disable_others=True, + ) + time.sleep(WARMUP_S) + + ndev, nmono = _start_capture(nplay) + hdev, hmono = _start_capture(hub2) + time.sleep(CAPTURE_S) + + # Snapshot while Hub2 is still alive (its callback stops when it closes). + ndev, nmono = list(ndev), list(nmono) + hdev, hmono = list(hdev), list(hmono) + + assert len(ndev) > 1000, f"No LSL-fed nPlay data flowing ({len(ndev)})" + assert len(hdev) > 1000, f"No Hub2 data flowing ({len(hdev)})" + + ns = _analyze(ndev, nmono) + hs = _analyze(hdev, hmono) + print( + f"\n=== LSL-nPlay: {ns['n_segments']} seg, {ns['n_backsteps']} backstep(s), " + f"max {ns['max_backstep_s'] * 1e3:.3f} ms | " + f"Hub2: {hs['n_segments']} seg, {hs['n_backsteps']} backstep(s), " + f"max {hs['max_backstep_s'] * 1e3:.3f} ms ===" + ) + + # Neither the LSL-origin clock nor Hub2's PTP clock wraps. + assert ns["n_wraps"] == 0 and hs["n_wraps"] == 0, ( + f"Unexpected device-clock reset(s): LSL-nPlay={ns['n_wraps']}, Hub2={hs['n_wraps']}" + ) + assert ns["n_segments"] > 0, "No evaluable LSL-nPlay timeline captured" + assert hs["n_segments"] > 0, "No evaluable Hub2 timeline captured" + + assert hs["n_backsteps"] == 0, ( + f"Live Hub2 host-mapped timeline went non-monotonic (max " + f"{hs['max_backstep_s'] * 1e3:.3f} ms) — unexpected for a real PTP clock; " + "check the rig before blaming nPlay." + ) + assert ns["n_backsteps"] == 0, ( + f"{ns['n_backsteps']} backward jump(s) in the LSL-fed nPlay host-mapped " + f"timeline (max {ns['max_backstep_s'] * 1e3:.3f} ms) while a real Hub2 " + "stayed monotonic — the cross-device merge stall reproduces on the LSL path." + ) diff --git a/src/cbdev/include/cbdev/clock_sync.h b/src/cbdev/include/cbdev/clock_sync.h index a7ddf574..cc28bc04 100644 --- a/src/cbdev/include/cbdev/clock_sync.h +++ b/src/cbdev/include/cbdev/clock_sync.h @@ -102,6 +102,14 @@ class ClockSync { /// Uncertainty (RTT/2) from the best probe. [[nodiscard]] std::optional getUncertaintyNs() const; + /// Monotonically-increasing counter, bumped whenever the committed offset + /// changes *discontinuously* (a step / re-acquire / source change), and + /// NEVER on a smooth slew or a sub-deadband converge. Consumers that + /// enforce post-conversion monotonicity reset their floor when this value + /// changes — it is the structural signal for "the converted timeline just + /// moved to a new regime, don't clamp across it." + [[nodiscard]] uint64_t syncEpoch() const; + /// True if probes are producing consistent offsets (spread < threshold). /// When false, the caller should feed data-packet timestamps as a fallback. [[nodiscard]] bool probesAreReliable() const; @@ -144,6 +152,7 @@ class ClockSync { std::optional m_current_offset_ns; std::optional m_current_uncertainty_ns; + uint64_t m_sync_epoch = 0; // bumped on discontinuous offset commits int m_pending_step_count = 0; // consecutive large-jump samples seen int m_nonconverged_streak = 0; // consecutive samples not within deadband bool m_committed_from_external = false; // committed value came from a peer offset diff --git a/src/cbdev/include/cbdev/device_session.h b/src/cbdev/include/cbdev/device_session.h index 6963db8e..48af307f 100644 --- a/src/cbdev/include/cbdev/device_session.h +++ b/src/cbdev/include/cbdev/device_session.h @@ -362,6 +362,11 @@ class IDeviceSession { /// @return Uncertainty in nanoseconds, or nullopt if no sync data available [[nodiscard]] virtual std::optional getUncertaintyNs() const = 0; + /// Discontinuity epoch — advances whenever the committed clock offset steps + /// to a new regime (re-acquire / step / wrap), never on a smooth slew. + /// Consumers reset post-conversion monotonic floors when this changes. + [[nodiscard]] virtual uint64_t syncEpoch() const = 0; + /// Inject an externally-determined offset (e.g., from a peer device). /// When set, overrides internal probe/data estimates in toLocalTime(). /// Pass nullopt to clear and revert to internal estimates. diff --git a/src/cbdev/src/clock_sync.cpp b/src/cbdev/src/clock_sync.cpp index 2b6885a5..97c6fc5c 100644 --- a/src/cbdev/src/clock_sync.cpp +++ b/src/cbdev/src/clock_sync.cpp @@ -121,6 +121,11 @@ std::optional ClockSync::getUncertaintyNs() const { return m_current_uncertainty_ns; } +uint64_t ClockSync::syncEpoch() const { + std::lock_guard lock(m_mutex); + return m_sync_epoch; +} + void ClockSync::setExternalOffset(std::optional offset_ns, std::optional uncertainty_ns) { std::lock_guard lock(m_mutex); @@ -267,6 +272,19 @@ bool ClockSync::externalPlausible(int64_t external_ns, void ClockSync::recomputeEstimate() { const InternalEstimate internal = computeInternalEstimate(); + // Snapshot the committed offset before we touch it, so the discontinuity + // epoch is bumped only when the committed value actually steps to a new + // regime — not when a stable external/peer offset is merely re-affirmed on + // every sample (which would churn the epoch and defeat the monotonic floor). + const std::optional prev_committed = m_current_offset_ns; + const auto stepped = [&](std::optional next) { + // Discontinuous = the value appears, disappears, or jumps beyond the + // converge deadband. Sub-deadband moves leave the timeline in place. + if (prev_committed.has_value() != next.has_value()) return true; + return prev_committed && next && + std::llabs(*next - *prev_committed) > m_config.commit_deadband_ns; + }; + // An external (peer-borrowed) offset is treated as a candidate, not an // unconditional override. It is adopted only when it is sanity-consistent // with our own evidence, and — because this runs on every probe/data @@ -277,6 +295,7 @@ void ClockSync::recomputeEstimate() { if (m_external_offset_ns && externalPlausible(*m_external_offset_ns, internal)) { // A plausible peer offset is authoritative (and already disciplined at // its source), so adopt it directly. + if (stepped(m_external_offset_ns)) ++m_sync_epoch; m_current_offset_ns = m_external_offset_ns; m_current_uncertainty_ns = m_external_uncertainty_ns ? m_external_uncertainty_ns @@ -291,14 +310,19 @@ void ClockSync::recomputeEstimate() { // deliberate event, not jitter — re-acquire directly. Only the same // (internal) source's drift is run through the commit discipline. if (m_committed_from_external || !m_current_offset_ns.has_value()) { + if (stepped(internal.offset_ns)) ++m_sync_epoch; m_current_offset_ns = *internal.offset_ns; m_current_uncertainty_ns = internal.uncertainty_ns; resetDiscipline(); } else { + // commitDisciplined() bumps the epoch itself on its discontinuous + // (cold-start / stepout / step-persist) paths, and leaves it + // untouched on a smooth slew or a sub-deadband converge. commitDisciplined(*internal.offset_ns, internal.uncertainty_ns); } m_committed_from_external = false; } else { + if (stepped(std::nullopt)) ++m_sync_epoch; m_current_offset_ns = std::nullopt; m_current_uncertainty_ns = std::nullopt; resetDiscipline(); @@ -314,6 +338,7 @@ void ClockSync::resetDiscipline() { void ClockSync::commitDisciplined(int64_t target_offset_ns, int64_t uncertainty_ns) { // Cold start: nothing committed yet — accept the target as-is. if (!m_current_offset_ns) { + ++m_sync_epoch; // first acquisition is a regime boundary m_current_offset_ns = target_offset_ns; m_current_uncertainty_ns = uncertainty_ns; resetDiscipline(); @@ -336,6 +361,7 @@ void ClockSync::commitDisciplined(int64_t target_offset_ns, int64_t uncertainty_ // that slew/step never reconciled (e.g. a lone device with no peer to // break the tie). if (++m_nonconverged_streak >= m_config.stepout_samples) { + ++m_sync_epoch; // abrupt re-acquire after a stuck offset m_current_offset_ns = target_offset_ns; m_current_uncertainty_ns = uncertainty_ns; resetDiscipline(); @@ -354,6 +380,7 @@ void ClockSync::commitDisciplined(int64_t target_offset_ns, int64_t uncertainty_ // Large jump: only accept once it persists, so a one-off outlier does // not move the committed offset. if (++m_pending_step_count >= m_config.step_persist) { + ++m_sync_epoch; // confirmed step applied abruptly m_current_offset_ns = target_offset_ns; m_current_uncertainty_ns = uncertainty_ns; resetDiscipline(); diff --git a/src/cbdev/src/device_session.cpp b/src/cbdev/src/device_session.cpp index 0eb30e92..10d13625 100644 --- a/src/cbdev/src/device_session.cpp +++ b/src/cbdev/src/device_session.cpp @@ -988,6 +988,11 @@ std::optional DeviceSession::getUncertaintyNs() const { return m_impl->clock_sync.getUncertaintyNs(); } +uint64_t DeviceSession::syncEpoch() const { + if (!m_impl) return 0; + return m_impl->clock_sync.syncEpoch(); +} + void DeviceSession::setExternalClockOffset(std::optional offset_ns, std::optional uncertainty_ns) { if (!m_impl) return; diff --git a/src/cbdev/src/device_session_impl.h b/src/cbdev/src/device_session_impl.h index d082490b..7eb505e2 100644 --- a/src/cbdev/src/device_session_impl.h +++ b/src/cbdev/src/device_session_impl.h @@ -311,6 +311,7 @@ class DeviceSession : public IDeviceSession { [[nodiscard]] std::optional getOffsetNs() const override; [[nodiscard]] std::optional getInternalOffsetNs() const override; [[nodiscard]] std::optional getUncertaintyNs() const override; + [[nodiscard]] uint64_t syncEpoch() const override; void setExternalClockOffset(std::optional offset_ns, std::optional uncertainty_ns = std::nullopt) override; diff --git a/src/cbdev/src/device_session_wrapper.h b/src/cbdev/src/device_session_wrapper.h index dadc5cff..2c073b55 100644 --- a/src/cbdev/src/device_session_wrapper.h +++ b/src/cbdev/src/device_session_wrapper.h @@ -234,6 +234,10 @@ class DeviceSessionWrapper : public IDeviceSession { return m_device.getUncertaintyNs(); } + [[nodiscard]] uint64_t syncEpoch() const override { + return m_device.syncEpoch(); + } + void setExternalClockOffset(std::optional offset_ns, std::optional uncertainty_ns = std::nullopt) override { m_device.setExternalClockOffset(offset_ns, uncertainty_ns); diff --git a/src/cbsdk/include/cbsdk/cbsdk.h b/src/cbsdk/include/cbsdk/cbsdk.h index 0ca0f21a..96673608 100644 --- a/src/cbsdk/include/cbsdk/cbsdk.h +++ b/src/cbsdk/include/cbsdk/cbsdk.h @@ -1122,6 +1122,39 @@ CBSDK_API cbsdk_result_t cbsdk_session_get_clock_uncertainty( /// @return CBSDK_RESULT_SUCCESS on success, error code on failure CBSDK_API cbsdk_result_t cbsdk_session_send_clock_probe(cbsdk_session_t session); +/// Convert a batch of device timestamps (nanoseconds) to host steady_clock +/// nanoseconds, optionally enforcing per-stream monotonicity. +/// +/// The whole batch is converted against one consistent clock snapshot. When +/// @p stream_id >= 0 the outputs are clamped to be non-decreasing relative to +/// the previous conversion request on that same stream_id, except across a +/// clock re-sync (where the floor resets so the timeline follows the new +/// regime). Pass @p stream_id < 0 for a stateless conversion (no clamping). +/// Monotonic streams assume in-order submission per stream_id; their state is +/// created lazily and freed by cbsdk_session_reset_monotonic() or session close. +/// +/// @param session Session handle (must not be NULL) +/// @param stream_id Opaque per-stream key (e.g. sample group); <0 = stateless +/// @param device_ns Input device timestamps, n entries (must not be NULL) +/// @param[out] out_steady_ns Output host steady_clock nanoseconds, n entries (must not be NULL) +/// @param n Number of timestamps +/// @return CBSDK_RESULT_SUCCESS if converted, CBSDK_RESULT_NOT_RUNNING if no sync data +CBSDK_API cbsdk_result_t cbsdk_session_to_local_time( + cbsdk_session_t session, + int64_t stream_id, + const uint64_t* device_ns, + int64_t* out_steady_ns, + size_t n); + +/// Drop the monotonic floor/epoch state for one stream_id used by +/// cbsdk_session_to_local_time(). No-op if the stream is unknown. +/// @param session Session handle (must not be NULL) +/// @param stream_id Stream key previously used for monotonic conversion +/// @return CBSDK_RESULT_SUCCESS, or CBSDK_RESULT_INVALID_PARAMETER on a bad handle +CBSDK_API cbsdk_result_t cbsdk_session_reset_monotonic( + cbsdk_session_t session, + int64_t stream_id); + /////////////////////////////////////////////////////////////////////////////////////////////////// // Utility /////////////////////////////////////////////////////////////////////////////////////////////////// diff --git a/src/cbsdk/include/cbsdk/sdk_session.h b/src/cbsdk/include/cbsdk/sdk_session.h index 089df375..1066f92c 100644 --- a/src/cbsdk/include/cbsdk/sdk_session.h +++ b/src/cbsdk/include/cbsdk/sdk_session.h @@ -769,6 +769,32 @@ class SdkSession { std::optional toDeviceTime(std::chrono::steady_clock::time_point local_time) const; + /// Convert a batch of device timestamps (nanoseconds) to host steady_clock + /// nanoseconds, optionally enforcing per-stream monotonicity. + /// + /// The whole batch is converted against one consistent (offset, epoch) + /// snapshot. When @p stream_id >= 0, each output is clamped to be + /// non-decreasing relative to the previous conversion *request* on that same + /// stream_id — except across a clock re-sync (discontinuity epoch change), + /// where the floor is reset so the timeline follows the new regime instead + /// of stalling. Pass @p stream_id < 0 for a stateless conversion (no + /// clamping). Monotonic streams assume in-order submission per stream_id; + /// state is created lazily and lives until resetMonotonic() or session end. + /// + /// @param stream_id Opaque per-stream key (e.g. sample group); <0 = stateless + /// @param device_ns Input device timestamps (n entries) + /// @param out_steady_ns Output host steady_clock nanoseconds (n entries) + /// @param n Number of timestamps + /// @return true on success; false if no clock sync data is available + bool toLocalTimeBatch(int64_t stream_id, + const uint64_t* device_ns, + int64_t* out_steady_ns, + size_t n) const; + + /// Drop the monotonic floor/epoch state for one stream_id (next monotonic + /// conversion on it starts fresh). No-op if the stream is unknown. + void resetMonotonic(int64_t stream_id); + /// Send a clock synchronization probe to the device. /// @return Result indicating success or error Result sendClockProbe(); diff --git a/src/cbsdk/src/cbsdk.cpp b/src/cbsdk/src/cbsdk.cpp index 99533dd5..7e7a9080 100644 --- a/src/cbsdk/src/cbsdk.cpp +++ b/src/cbsdk/src/cbsdk.cpp @@ -1667,6 +1667,38 @@ cbsdk_result_t cbsdk_session_send_clock_probe(cbsdk_session_t session) { } } +cbsdk_result_t cbsdk_session_to_local_time( + cbsdk_session_t session, + int64_t stream_id, + const uint64_t* device_ns, + int64_t* out_steady_ns, + size_t n) { + if (!session || !session->cpp_session || !device_ns || !out_steady_ns) { + return CBSDK_RESULT_INVALID_PARAMETER; + } + try { + const bool ok = session->cpp_session->toLocalTimeBatch( + stream_id, device_ns, out_steady_ns, n); + return ok ? CBSDK_RESULT_SUCCESS : CBSDK_RESULT_NOT_RUNNING; + } catch (...) { + return CBSDK_RESULT_INTERNAL_ERROR; + } +} + +cbsdk_result_t cbsdk_session_reset_monotonic( + cbsdk_session_t session, + int64_t stream_id) { + if (!session || !session->cpp_session) { + return CBSDK_RESULT_INVALID_PARAMETER; + } + try { + session->cpp_session->resetMonotonic(stream_id); + return CBSDK_RESULT_SUCCESS; + } catch (...) { + return CBSDK_RESULT_INTERNAL_ERROR; + } +} + /////////////////////////////////////////////////////////////////////////////////////////////////// // Utility /////////////////////////////////////////////////////////////////////////////////////////////////// diff --git a/src/cbsdk/src/sdk_session.cpp b/src/cbsdk/src/sdk_session.cpp index e67c25fe..19256fc0 100644 --- a/src/cbsdk/src/sdk_session.cpp +++ b/src/cbsdk/src/sdk_session.cpp @@ -32,6 +32,8 @@ #include #include #include +#include +#include #include "cbdev/clock_sync.h" #ifndef _WIN32 #include @@ -332,6 +334,52 @@ struct SdkSession::Impl { PendingClockProbe pending_clock_probe; std::mutex clock_probe_mutex; + // Per-stream monotonic-conversion state (see toLocalTimeBatch). Each stream + // keeps its own non-decreasing floor and the discontinuity epoch it last + // saw; on an epoch change the floor is reset (a genuine clock re-sync), else + // a backward step is clamped to the floor. Lazily created per stream_id. + struct MonoState { + int64_t floor_ns = 0; + uint64_t last_epoch = 0; + bool seen = false; + }; + std::mutex mono_mutex; + std::unordered_map mono_streams; + + // CLIENT-mode (shmem) discontinuity-epoch approximation. The shmem offset + // path has no local ClockSync, so we derive an epoch by watching the + // peer/Central offset for jumps larger than a smooth slew (Q1=b — best + // effort until the shmem layout carries a real epoch field). Guarded by + // mono_mutex (only touched from the monotonic conversion path). + std::optional client_epoch_offset; + uint64_t client_epoch = 0; + + uint64_t deriveClientEpoch(int64_t offset_ns) { // mono_mutex held + constexpr int64_t kStepNs = 50'000'000; // ~ClockSync slew_max_ns + if (client_epoch_offset && + std::llabs(offset_ns - *client_epoch_offset) > kStepNs) + ++client_epoch; + client_epoch_offset = offset_ns; + return client_epoch; + } + + // Current (offset, epoch) from whichever source getClockOffsetNs() would + // use, read together so a batch converts against one consistent regime. + std::optional> currentOffsetAndEpoch() { // mono_mutex held + if (device_session) { + auto off = device_session->getOffsetNs(); + if (off) return std::make_pair(*off, device_session->syncEpoch()); + return std::nullopt; + } + if (shmem_session) { + auto off = shmem_session->getClockOffsetNs(); + if (off) return std::make_pair(*off, deriveClientEpoch(*off)); + } + auto off = client_clock_sync.getOffsetNs(); + if (off) return std::make_pair(*off, client_clock_sync.syncEpoch()); + return std::nullopt; + } + // Statistics — atomic counters, no mutex needed (Phase 2, Fix 9) struct AtomicStats { std::atomic packets_received_from_device{0}; @@ -2404,6 +2452,51 @@ SdkSession::toLocalTime(uint64_t device_time_ns) const { return std::nullopt; } +bool SdkSession::toLocalTimeBatch(int64_t stream_id, + const uint64_t* device_ns, + int64_t* out_steady_ns, + size_t n) const { + if (n == 0) return true; + if (!device_ns || !out_steady_ns) return false; + + // Stateless fast path: one offset, plain subtraction, no per-stream state. + if (stream_id < 0) { + const auto offset = getClockOffsetNs(); + if (!offset) return false; + for (size_t i = 0; i < n; ++i) + out_steady_ns[i] = static_cast(device_ns[i]) - *offset; + return true; + } + + // Monotonic path: take a single (offset, epoch) snapshot and the floor map + // under one lock so the whole batch sees one consistent clock regime. + std::lock_guard lock(m_impl->mono_mutex); + const auto oe = m_impl->currentOffsetAndEpoch(); + if (!oe) return false; + const int64_t offset = oe->first; + const uint64_t epoch = oe->second; + + auto& st = m_impl->mono_streams[stream_id]; // lazy-create (seen == false) + for (size_t i = 0; i < n; ++i) { + const int64_t raw = static_cast(device_ns[i]) - offset; + // Reset the floor on the first conversion for this stream or whenever + // the clock regime changed; otherwise clamp to a non-decreasing floor. + const int64_t out = (!st.seen || epoch != st.last_epoch) + ? raw + : std::max(raw, st.floor_ns); + st.floor_ns = out; + st.last_epoch = epoch; + st.seen = true; + out_steady_ns[i] = out; + } + return true; +} + +void SdkSession::resetMonotonic(int64_t stream_id) { + std::lock_guard lock(m_impl->mono_mutex); + m_impl->mono_streams.erase(stream_id); +} + std::optional SdkSession::toDeviceTime(std::chrono::steady_clock::time_point local_time) const { // STANDALONE: delegate to device_session's ClockSync diff --git a/tests/integration/CMakeLists.txt b/tests/integration/CMakeLists.txt index 54507e4d..7f1384a6 100644 --- a/tests/integration/CMakeLists.txt +++ b/tests/integration/CMakeLists.txt @@ -21,7 +21,7 @@ include(GoogleTest) # Download nPlayServer binary and test data from the CereLink GitHub release ############################################################################### -set(CERELINK_RELEASE_URL "https://github.com/CerebusOSS/CereLink/releases/download/v9.3.0") +set(CERELINK_RELEASE_URL "https://github.com/CerebusOSS/CereLink/releases/download/v9.11.0") set(NPLAY_CACHE_DIR "${CMAKE_CURRENT_BINARY_DIR}/_nplay_cache") file(MAKE_DIRECTORY ${NPLAY_CACHE_DIR}) @@ -147,6 +147,7 @@ if(NPLAY_AVAILABLE AND NPLAY_DATA_AVAILABLE) test_sdk_data_reception.cpp test_sdk_clock_sync.cpp test_multidevice_clock_sync.cpp + test_nplay_sync_stress.cpp ) target_link_libraries(cbsdk_integration_tests diff --git a/tests/integration/nplay_fixture.h b/tests/integration/nplay_fixture.h index 0b14e60f..172ec754 100644 --- a/tests/integration/nplay_fixture.h +++ b/tests/integration/nplay_fixture.h @@ -15,6 +15,7 @@ #include +#include #include #include #include @@ -50,13 +51,25 @@ class NPlayFixture : public ::testing::Test { protected: void SetUp() override { + // Unique lock name PER LAUNCH, not per process: gtest runs every TEST_F + // in one process, so a pid-only name would collide across consecutive + // SetUp()s. The trailing counter guarantees each nPlayServer instance + // gets its own lock (and its own plugin-container shmem channel names, + // which are also derived from the lock name). m_lockName = "nplay_integ_" + std::to_string( #ifdef _WIN32 GetCurrentProcessId() #else getpid() #endif - ); + ) + "_" + std::to_string(nextLaunchId()); + +#ifndef _WIN32 + // Defensive: clear any stale lock file a hard-killed prior run left + // behind. nPlayServer guards the lock with O_EXCL, so a leftover file + // would make this launch fail with "Cannot run multiple instances". + std::remove(lockFilePath().c_str()); +#endif std::string binary = NPLAY_BINARY_PATH; std::string ns6 = NPLAY_NS6_PATH; @@ -112,9 +125,15 @@ class NPlayFixture : public ::testing::Test { } m_pid = -1; - // Clean up the POSIX named semaphore - std::string sem_name = "/" + m_lockName; - sem_unlink(sem_name.c_str()); + // nPlayServer's lock is a FILE ($TMPDIR/.lock guarded by + // O_EXCL), not a named semaphore. A SIGKILLed instance leaves it + // behind, so remove it here. (The per-launch unique name already + // prevents collisions; this just keeps TMPDIR from accumulating + // stale locks.) + std::remove(lockFilePath().c_str()); + // Belt-and-suspenders: older builds also created a named semaphore; + // unlinking a non-existent name is harmless. + sem_unlink(("/" + m_lockName).c_str()); } #endif } @@ -147,6 +166,22 @@ class NPlayFixture : public ::testing::Test { } private: + /// Monotonically increasing id so each launch gets a unique lock name, + /// even though all TEST_F instances share one process (and thus one pid). + static unsigned nextLaunchId() { + static std::atomic counter{0}; + return counter.fetch_add(1, std::memory_order_relaxed); + } + + /// Path of nPlayServer's lock file for the current lock name, computed the + /// same way nPlayServer does ($TMPDIR/.lock, falling back to /tmp). + std::string lockFilePath() const { + const char* tmp = std::getenv("TMPDIR"); + std::string dir = (tmp != nullptr && tmp[0] != '\0') ? tmp : "/tmp"; + if (!dir.empty() && dir.back() == '/') dir.pop_back(); + return dir + "/" + m_lockName + ".lock"; + } + std::string m_lockName; #ifdef _WIN32 PROCESS_INFORMATION m_pi = {}; diff --git a/tests/integration/test_nplay_sync_stress.cpp b/tests/integration/test_nplay_sync_stress.cpp new file mode 100644 index 00000000..e377aa5b --- /dev/null +++ b/tests/integration/test_nplay_sync_stress.cpp @@ -0,0 +1,402 @@ +/////////////////////////////////////////////////////////////////////////////////////////////////// +/// @file test_nplay_sync_stress.cpp +/// @author CereLink Development Team +/// +/// @brief Stress tests for nPlayServer clock synchronization +/// +/// nPlayServer replays a recording (or an LSL stream) and stamps each packet +/// with a DEVICE clock that is NOT a real wall clock: for file playback it is +/// "samples since the start of the file", which RESETS to ~0 every time the +/// short test file wraps. CereLink's ClockSync maps that device clock onto the +/// host steady_clock; downstream consumers (e.g. ezmsg-blackrock RESAMPLE/MERGE) +/// then require the resulting HOST-mapped timeline to be MONOTONIC. +/// +/// The failure mode these tests guard against (observed in the cursor pipeline, +/// 2026-06): nPlay's host-mapped timestamps jump BACKWARD by ~0.6 s *within* a +/// single playback pass — the offset estimate recalibrates against a replay +/// clock that does not advance at true wall-rate — which stalls any consumer +/// that assumes monotonic time. A real Hub on a stable PTP clock does not do +/// this, so the symptom only appears when nPlay feeds a monotonic-time consumer. +/// +/// Two scenarios: +/// 1. SINGLE nPlay (always runs when the nPlayServer binary + test data are +/// available, like the other integration tests): stamp every data batch +/// with toLocalTime(first_ts) — exactly as a consumer does — over a long +/// run that crosses many file wraps, then assert the host-mapped timeline +/// is monotonic WITHIN each no-wrap segment (the wrap itself is a legitimate +/// device-clock reset and is excluded). +/// +/// 2. nPlay + a LIVE Hub2 (hardware-gated, opt-in): the actual cross-device +/// merge topology. Brings up an nPlay session AND a real Hub2 session +/// concurrently and asserts BOTH host-mapped timelines stay monotonic and +/// that each device's current time maps to ~now() — so nPlay's replay clock +/// cannot silently diverge from the real Hub while a merge consumes both. +/// SKIPPED unless CERELINK_HW_MULTIDEVICE is set (bringing up a live Hub +/// runs a handshake that can HARDRESET it — disruptive on a recording rig). +/// +/// The wrap-aware, segmented monotonicity check is intentional: a naive "offset +/// is constant" assertion would false-positive at every file wrap, where the +/// committed offset MUST step by the file length to keep current device time +/// mapped to ~now. +/////////////////////////////////////////////////////////////////////////////////////////////////// + +#include +#include "nplay_fixture.h" + +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace cbsdk; + +namespace { + +/// Channels to stream at 30 kHz while measuring the timeline. +constexpr uint32_t N_CHANS = 8; + +/// Seconds to wait after configuring channels for the first clock probe / data +/// fallback to produce an offset (the SDK probes at handshake then every ~2 s). +constexpr int SYNC_WARMUP_S = 3; + +/// A backward jump in DEVICE time larger than this marks a file wrap (the replay +/// counter resetting to ~0), not a glitch. The dnss256 test file is ~2 s, so a +/// wrap drops device time by ~2e9 ns; a within-pass sample step is ~3e4 ns. +constexpr int64_t WRAP_BACK_NS = 50'000'000; // 50 ms + +/// After a wrap the committed offset must STEP to track the reset device clock; +/// the commit discipline (step_persist / stepout_samples) takes a few samples to +/// re-acquire. Ignore host-mapped samples within this much DEVICE time of the +/// start of a segment so the legitimate re-acquisition is not counted as a bug. +constexpr int64_t POST_WRAP_GUARD_NS = 150'000'000; // 150 ms + +/// Maximum tolerated BACKWARD jump in the host-mapped timeline within a no-wrap +/// segment. Healthy localhost sync keeps the offset stable to well under a +/// millisecond, so the host-mapped time tracks device time ~1:1 and never steps +/// back. The guarded-against bug jumps ~600 ms; this threshold sits ~30x below +/// that while tolerating normal sub-deadband offset corrections. +constexpr int64_t MONO_BACKSTEP_NS = 20'000'000; // 20 ms + +/// Only evaluate segments whose post-guard device span exceeds this, so a +/// partial segment captured at the very start/end of the run is not judged. +constexpr int64_t MIN_SEGMENT_SPAN_NS = 300'000'000; // 300 ms + +/// One captured data batch: the first sample's device time and the host time it +/// mapped to, computed live (in the callback) just as a consumer would stamp it. +struct TimelineSample { + uint64_t dev_ns; ///< first device timestamp in the batch (nanoseconds) + int64_t host_ns; ///< toLocalTime(dev_ns) as ns since the steady epoch + bool host_valid; ///< false if no clock offset was available yet +}; + +/// Result of walking a captured timeline and checking within-segment monotonicity. +struct TimelineStats { + size_t n_samples = 0; ///< valid (synced) samples evaluated + size_t n_wraps = 0; ///< device-clock resets detected + size_t n_segments = 0; ///< no-wrap segments long enough to evaluate + size_t n_backsteps = 0; ///< within-segment backward jumps > MONO_BACKSTEP_NS + int64_t max_backstep_ns = 0;///< largest within-segment backward jump seen (>=0) +}; + +/// Capture a continuous device→host timeline by stamping each 30 kHz data batch +/// with toLocalTime(first_ts), the way an acquisition consumer does. +class TimelineCapture { +public: + /// Register the batch callback on @p session. Stamps run on the callback + /// thread; both conversion paths are internally locked, so this is safe. + /// + /// @param stream_id <0 (default): stamp with the stateless toLocalTime(), + /// the raw mapping a naive consumer uses. >=0: stamp with the monotonic + /// toLocalTimeBatch() on that stream_id, which enforces a non-decreasing + /// host timeline (resetting only across a clock re-sync / file wrap). + explicit TimelineCapture(SdkSession& session, int64_t stream_id = -1) { + m_handle = session.registerGroupBatchCallback( + SampleRate::SR_30kHz, + [this, &session, stream_id](const int16_t*, size_t n_samples, size_t, + const uint64_t* timestamps) { + if (n_samples == 0) return; + const uint64_t dev_ns = timestamps[0]; + TimelineSample s; + s.dev_ns = dev_ns; + if (stream_id < 0) { + auto local = session.toLocalTime(dev_ns); + s.host_valid = local.has_value(); + s.host_ns = local.has_value() + ? std::chrono::duration_cast( + local->time_since_epoch()).count() + : 0; + } else { + int64_t out = 0; + s.host_valid = + session.toLocalTimeBatch(stream_id, &dev_ns, &out, 1); + s.host_ns = s.host_valid ? out : 0; + } + std::lock_guard lock(m_mutex); + m_samples.push_back(s); + }); + } + + std::vector drain() { + std::lock_guard lock(m_mutex); + return m_samples; + } + +private: + std::mutex m_mutex; + std::vector m_samples; + CallbackHandle m_handle = 0; +}; + +/// Evaluate one no-wrap segment [begin, end) and fold its findings into @p st. +/// (Defined below; forward-declared so analyzeTimeline can call it.) +void analyzeSegment(const std::vector& samples, + size_t begin, size_t end, TimelineStats& st, + int64_t backstep_threshold); + +/// Walk a captured timeline: split it into no-wrap segments (a wrap is a large +/// backward jump in DEVICE time), skip a short re-acquisition guard at the start +/// of each segment, and count BACKWARD jumps in the host-mapped time that exceed +/// @p backstep_threshold. Healthy sync produces zero. The raw mapping uses a +/// tolerant threshold (sub-deadband corrections are allowed); the monotonic +/// mapping is checked with threshold 0 (it must never step back at all). +TimelineStats analyzeTimeline(const std::vector& samples, + int64_t backstep_threshold = MONO_BACKSTEP_NS) { + TimelineStats st; + + // First pass: locate segment boundaries from the device-time series alone, + // independent of whether the offset was available at that instant. + size_t seg_start = 0; + for (size_t i = 0; i < samples.size(); ++i) { + const bool wrap = i > 0 && + static_cast(samples[i].dev_ns) - + static_cast(samples[i - 1].dev_ns) < -WRAP_BACK_NS; + if (wrap) { + ++st.n_wraps; + analyzeSegment(samples, seg_start, i, st, backstep_threshold); + seg_start = i; + } + } + analyzeSegment(samples, seg_start, samples.size(), st, backstep_threshold); + return st; +} + +void analyzeSegment(const std::vector& samples, + size_t begin, size_t end, TimelineStats& st, + int64_t backstep_threshold) { + if (end <= begin) return; + const uint64_t seg_dev_start = samples[begin].dev_ns; + + // Span check uses the guarded portion only. + bool have_prev = false; + int64_t prev_host = 0; + int64_t seg_first_eval = -1, seg_last_eval = -1; + size_t seg_samples = 0, seg_backsteps = 0; + int64_t seg_max_backstep = 0; + + for (size_t i = begin; i < end; ++i) { + const auto& s = samples[i]; + if (!s.host_valid) { have_prev = false; continue; } + // Skip the post-wrap re-acquisition window. + if (static_cast(s.dev_ns - seg_dev_start) < POST_WRAP_GUARD_NS) { + have_prev = false; + continue; + } + if (seg_first_eval < 0) seg_first_eval = static_cast(s.dev_ns); + seg_last_eval = static_cast(s.dev_ns); + ++seg_samples; + + if (have_prev) { + const int64_t backstep = prev_host - s.host_ns; // >0 = went backward + if (backstep > seg_max_backstep) seg_max_backstep = backstep; + if (backstep > backstep_threshold) ++seg_backsteps; + } + prev_host = s.host_ns; + have_prev = true; + } + + // Ignore segments too short (post-guard) to judge meaningfully. + if (seg_first_eval < 0 || (seg_last_eval - seg_first_eval) < MIN_SEGMENT_SPAN_NS) + return; + + ++st.n_segments; + st.n_samples += seg_samples; + st.n_backsteps += seg_backsteps; + if (seg_max_backstep > st.max_backstep_ns) st.max_backstep_ns = seg_max_backstep; +} + +Result createSession(DeviceType type) { + SdkConfig config; + config.device_type = type; + return SdkSession::create(config); +} + +bool hardwareHub2Enabled() { + const char* v = std::getenv("CERELINK_HW_MULTIDEVICE"); + return v != nullptr && v[0] != '\0'; +} + +} // namespace + +/////////////////////////////////////////////////////////////////////////////////////////////////// +// Scenario 1: single nPlay — host-mapped timeline monotonic across a long run +/////////////////////////////////////////////////////////////////////////////////////////////////// + +class NPlaySyncStressTest : public NPlayFixture {}; + +// Stamp every 30 kHz batch with toLocalTime(first_ts) over ~20 s of playback +// (many file wraps) and require the host-mapped timeline to be monotonic within +// each no-wrap segment. This is the direct regression guard for the ±0.6 s +// backward jumps that stall downstream RESAMPLE/MERGE. +TEST_F(NPlaySyncStressTest, HostMappedTimelineMonotonicAcrossLongPlayback) { + auto result = createSession(DeviceType::NPLAY); + ASSERT_TRUE(result.isOk()) << result.error(); + auto& session = result.value(); + + ASSERT_TRUE(session.setSampleGroup(N_CHANS, ChannelType::FRONTEND, + SampleRate::SR_30kHz, true).isOk()); + // Let channels configure and the first offset settle before capturing. + std::this_thread::sleep_for(std::chrono::seconds(SYNC_WARMUP_S)); + + TimelineCapture capture(session); + std::this_thread::sleep_for(std::chrono::seconds(20)); + session.stop(); + + const auto samples = capture.drain(); + ASSERT_GT(samples.size(), 1000u) + << "Too few batches captured (" << samples.size() << ") — no data flowing?"; + + const TimelineStats st = analyzeTimeline(samples); + + std::printf("=== nPlay timeline: %zu batches, %zu wraps, %zu segments, " + "%zu backstep(s) > %lld ms, max backstep = %.3f ms ===\n", + samples.size(), st.n_wraps, st.n_segments, st.n_backsteps, + static_cast(MONO_BACKSTEP_NS / 1'000'000), + static_cast(st.max_backstep_ns) / 1e6); + std::fflush(stdout); + + // Sanity: the run must actually exercise wraps and at least one real segment, + // otherwise the monotonicity assertion below is vacuous. + EXPECT_GT(st.n_wraps, 0u) << "Expected the short test file to wrap during 20 s"; + ASSERT_GT(st.n_segments, 0u) << "No evaluable no-wrap segment captured"; + + EXPECT_EQ(st.n_backsteps, 0u) + << st.n_backsteps << " within-pass backward jump(s) in the host-mapped " + << "timeline (max " << static_cast(st.max_backstep_ns) / 1e6 + << " ms). nPlay's offset is recalibrating against the replay clock — " + << "this is the non-monotonic-timestamp bug that stalls RESAMPLE/MERGE."; +} + +// Same long playback, but stamped through the MONOTONIC batch API +// (toLocalTimeBatch with a stream_id). Unlike the raw mapping above — which is +// only monotonic WITHIN a no-wrap segment and tolerates sub-deadband wiggle — +// the monotonic stream must NEVER step backward within a segment (threshold 0). +// At each file wrap the discontinuity epoch advances and the floor resets, so a +// single backward step per wrap is expected and excluded by the segmentation. +TEST_F(NPlaySyncStressTest, MonotonicStreamNeverStepsBackWithinSegment) { + auto result = createSession(DeviceType::NPLAY); + ASSERT_TRUE(result.isOk()) << result.error(); + auto& session = result.value(); + + ASSERT_TRUE(session.setSampleGroup(N_CHANS, ChannelType::FRONTEND, + SampleRate::SR_30kHz, true).isOk()); + std::this_thread::sleep_for(std::chrono::seconds(SYNC_WARMUP_S)); + + constexpr int64_t kMonoStream = 1; // 30 kHz raw stream + TimelineCapture capture(session, kMonoStream); + std::this_thread::sleep_for(std::chrono::seconds(20)); + session.stop(); + + const auto samples = capture.drain(); + ASSERT_GT(samples.size(), 1000u) + << "Too few batches captured (" << samples.size() << ") — no data flowing?"; + + // Threshold 0: the monotonic timeline must be exactly non-decreasing. + const TimelineStats st = analyzeTimeline(samples, /*backstep_threshold=*/0); + + std::printf("=== nPlay MONOTONIC timeline: %zu batches, %zu wraps, %zu segments, " + "%zu backstep(s), max backstep = %.6f ms ===\n", + samples.size(), st.n_wraps, st.n_segments, st.n_backsteps, + static_cast(st.max_backstep_ns) / 1e6); + std::fflush(stdout); + + EXPECT_GT(st.n_wraps, 0u) << "Expected the short test file to wrap during 20 s"; + ASSERT_GT(st.n_segments, 0u) << "No evaluable no-wrap segment captured"; + + EXPECT_EQ(st.n_backsteps, 0u) + << st.n_backsteps << " backward step(s) in the MONOTONIC host timeline " + << "(max " << static_cast(st.max_backstep_ns) / 1e6 << " ms). " + << "toLocalTimeBatch(stream_id) must clamp every within-segment backstep."; +} + +/////////////////////////////////////////////////////////////////////////////////////////////////// +// Scenario 2: nPlay + a LIVE Hub2 — both timelines stay monotonic and host-anchored +/////////////////////////////////////////////////////////////////////////////////////////////////// + +// The real cross-device merge topology: nPlay (first/reference device) plus a +// real Hub2. If nPlay's host-mapped timeline goes non-monotonic while Hub2's +// stays clean, a consumer resampling one onto the other stalls. Hardware-gated. +TEST_F(NPlaySyncStressTest, NPlayAndLiveHub2TimelinesStayMonotonic) { + if (!hardwareHub2Enabled()) { + GTEST_SKIP() << "Set CERELINK_HW_MULTIDEVICE=1 to run against a live Hub2 " + "alongside nPlay (the handshake may HARDRESET the Hub)."; + } + + auto nplay = createSession(DeviceType::NPLAY); + ASSERT_TRUE(nplay.isOk()) << "nPlay: " << nplay.error(); + auto hub2 = createSession(DeviceType::HUB2); + ASSERT_TRUE(hub2.isOk()) << "Hub2: " << hub2.error(); + + auto& nplay_s = nplay.value(); + auto& hub2_s = hub2.value(); + + ASSERT_TRUE(nplay_s.setSampleGroup(N_CHANS, ChannelType::FRONTEND, + SampleRate::SR_30kHz, true).isOk()); + ASSERT_TRUE(hub2_s.setSampleGroup(N_CHANS, ChannelType::FRONTEND, + SampleRate::SR_30kHz, true).isOk()); + std::this_thread::sleep_for(std::chrono::seconds(SYNC_WARMUP_S)); + + TimelineCapture nplay_cap(nplay_s); + TimelineCapture hub2_cap(hub2_s); + std::this_thread::sleep_for(std::chrono::seconds(15)); + nplay_s.stop(); + hub2_s.stop(); + + const auto nplay_samples = nplay_cap.drain(); + const auto hub2_samples = hub2_cap.drain(); + ASSERT_GT(nplay_samples.size(), 1000u) << "No nPlay data flowing"; + ASSERT_GT(hub2_samples.size(), 1000u) << "No Hub2 data flowing"; + + const TimelineStats ns = analyzeTimeline(nplay_samples); + const TimelineStats hs = analyzeTimeline(hub2_samples); + + std::printf("=== nPlay: %zu seg, %zu backstep(s), max %.3f ms | " + "Hub2: %zu seg, %zu backstep(s), max %.3f ms ===\n", + ns.n_segments, ns.n_backsteps, + static_cast(ns.max_backstep_ns) / 1e6, + hs.n_segments, hs.n_backsteps, + static_cast(hs.max_backstep_ns) / 1e6); + std::fflush(stdout); + + ASSERT_GT(ns.n_segments, 0u) << "No evaluable nPlay segment"; + ASSERT_GT(hs.n_segments, 0u) << "No evaluable Hub2 segment"; + + EXPECT_EQ(hs.n_backsteps, 0u) + << "Live Hub2 host-mapped timeline went non-monotonic (max " + << static_cast(hs.max_backstep_ns) / 1e6 << " ms) — unexpected " + << "for a real PTP clock; check the rig before blaming nPlay."; + EXPECT_EQ(ns.n_backsteps, 0u) + << ns.n_backsteps << " within-pass backward jump(s) in nPlay's host-mapped " + << "timeline (max " << static_cast(ns.max_backstep_ns) / 1e6 + << " ms) while a real Hub2 stayed monotonic — the cross-device merge " + << "stall reproduces here."; +} diff --git a/tests/unit/test_clock_sync.cpp b/tests/unit/test_clock_sync.cpp index 78db3f11..e9f38aa0 100644 --- a/tests/unit/test_clock_sync.cpp +++ b/tests/unit/test_clock_sync.cpp @@ -559,3 +559,114 @@ TEST(DeviceTimestampToNs, GeminiNanosecondsPassThrough) { TEST(DeviceTimestampToNs, TrivialDenominatorDisablesConversion) { EXPECT_EQ(deviceTimestampToNs(12345, /*ts_are_ns=*/false, 1, 1), 12345ULL); } + +/////////////////////////////////////////////////////////////////////////////////////////////////// +// Discontinuity epoch (syncEpoch) +// +// syncEpoch() must bump exactly when the committed offset steps to a new regime +// (cold-start, confirmed step, stepout, external adopt/revert, device wrap) and +// must NOT bump on a smooth slew, a sub-deadband converge, or while a stable +// external offset is merely re-affirmed every sample. Consumers reset their +// post-conversion monotonic floor on a change here, so churn here would defeat +// monotonicity and a missed bump would clamp across a real re-sync. +/////////////////////////////////////////////////////////////////////////////////////////////////// + +namespace { + +// Add a probe at host instant host_ns whose computed offset equals target_off. +// Fixed RTT = 1000 ns: offset = T3 - T1 - 0.5*RTT = T3 - host_ns - 500. +void addOffsetProbe(ClockSync& sync, int64_t host_ns, int64_t target_off) { + const uint64_t t3 = static_cast(target_off + host_ns + 500); + sync.addProbeSample(tp_from_ns(host_ns), t3, tp_from_ns(host_ns + 1000)); +} + +} // anonymous namespace + +// Starts at 0, and the first acquisition is itself a regime boundary. +TEST(ClockSyncEpochTest, ColdStartBumpsOnceFromZero) { + ClockSync sync; + EXPECT_EQ(sync.syncEpoch(), 0u); + + addOffsetProbe(sync, HOST_NOW_NS, TRUE_OFFSET_NS); + EXPECT_EQ(sync.syncEpoch(), 1u); +} + +// A stable offset (repeated within-deadband probes) must not churn the epoch. +TEST(ClockSyncEpochTest, StableOffsetDoesNotBump) { + ClockSync sync; + addOffsetProbe(sync, HOST_NOW_NS, TRUE_OFFSET_NS); + const uint64_t after_cold = sync.syncEpoch(); + + // Sub-deadband jitter (< commit_deadband_ns = 1 ms) around the same offset. + for (int k = 1; k < 12; ++k) { + const int64_t jitter = (k % 2 ? 1 : -1) * 200'000LL; // ±0.2 ms + addOffsetProbe(sync, HOST_NOW_NS + k * 1'000'000LL, TRUE_OFFSET_NS + jitter); + } + EXPECT_EQ(sync.syncEpoch(), after_cold) << "sub-deadband jitter churned the epoch"; +} + +// A confirmed step (> slew_max, persisted step_persist samples) bumps exactly +// once, and holding at the new level afterward does not bump again. +TEST(ClockSyncEpochTest, ConfirmedStepBumpsOnce) { + ClockSync sync; + addOffsetProbe(sync, HOST_NOW_NS, TRUE_OFFSET_NS); + const uint64_t before = sync.syncEpoch(); + + // Step the target +200 ms (> slew_max_ns = 50 ms). The first stepped probe + // is a lone upward outlier the glitch filter strips, so the step only starts + // counting once it is corroborated and then commits after step_persist (=3) + // consecutive samples. Feed enough to guarantee a single commit. + const int64_t stepped = TRUE_OFFSET_NS + 200'000'000LL; + for (int k = 1; k <= 8; ++k) + addOffsetProbe(sync, HOST_NOW_NS + k * 1'000'000LL, stepped); + EXPECT_EQ(sync.syncEpoch(), before + 1) << "confirmed step did not bump exactly once"; + + // Hold at the new level: now converged, must not bump. + for (int k = 9; k < 15; ++k) + addOffsetProbe(sync, HOST_NOW_NS + k * 1'000'000LL, stepped); + EXPECT_EQ(sync.syncEpoch(), before + 1) << "holding at the new level re-bumped"; +} + +// Adopting a plausible external offset bumps once; re-affirming the same external +// value on subsequent samples must NOT churn; reverting to internal bumps again. +TEST(ClockSyncEpochTest, ExternalAdoptReaffirmRevert) { + ClockSync sync; + for (int k = 0; k < 8; ++k) { + const int64_t recv_ns = HOST_NOW_NS + k * 1'000'000LL; + const uint64_t device_ns = + static_cast(recv_ns + TRUE_OFFSET_NS - 300'000); + sync.addDataPacketSample(device_ns, tp_from_ns(recv_ns)); + } + const uint64_t before_ext = sync.syncEpoch(); + + // Adopt a plausible external offset 50 ms away — one regime boundary. + sync.setExternalOffset(TRUE_OFFSET_NS + 50'000'000LL, 1'000'000); + const uint64_t after_adopt = sync.syncEpoch(); + EXPECT_EQ(after_adopt, before_ext + 1); + + // Re-affirm the same external value over many samples — no churn. + for (int k = 8; k < 24; ++k) { + const int64_t recv_ns = HOST_NOW_NS + k * 1'000'000LL; + const uint64_t device_ns = + static_cast(recv_ns + TRUE_OFFSET_NS - 300'000); + sync.addDataPacketSample(device_ns, tp_from_ns(recv_ns)); + } + EXPECT_EQ(sync.syncEpoch(), after_adopt) << "stable external offset churned the epoch"; + + // Revert to the internal estimate (50 ms away) — another regime boundary. + sync.setExternalOffset(std::nullopt); + EXPECT_EQ(sync.syncEpoch(), after_adopt + 1) << "reverting source did not bump"; +} + +// A device-clock wrap (offset jumps > 1 s, e.g. nPlay file loop) drops history +// and re-acquires — the epoch must advance so consumers reset their floor. +TEST(ClockSyncEpochTest, DeviceWrapBumps) { + ClockSync sync; + for (int k = 0; k < 4; ++k) + addOffsetProbe(sync, HOST_NOW_NS + k * 1'000'000LL, TRUE_OFFSET_NS); + const uint64_t before_wrap = sync.syncEpoch(); + + // Device timestamp jumps backward by 2 s relative to host -> offset drops 2 s. + addOffsetProbe(sync, HOST_NOW_NS + 5'000'000LL, TRUE_OFFSET_NS - 2'000'000'000LL); + EXPECT_GT(sync.syncEpoch(), before_wrap) << "device wrap did not advance the epoch"; +}