Fix wedged-acquisition freezes and prevent USB suspend from killing the stream

scripts/audio_latency_test/README.md

@@ -65,8 +65,32 @@ run_test.py --labrecorder-cli /path/to/LabRecorderCLI
 
 # Re-analyze an existing recording (no hardware):
 run_test.py --analyze-xdf recordings/audio_latency_xxx.xdf
+
+# Verify timestamps and exit non-zero if the MCC stream is bad:
+run_test.py --analyze-xdf recordings/audio_latency_xxx.xdf --fail-on-bad-timestamps
 ```
 
+## Timestamp verification
+
+Every run (live or `--analyze-xdf`) prints a **timestamp health check** for the
+MCC stream. Unlike the latency analysis — which loads with
+`dejitter_timestamps=True` and so *cleans up* the timestamps — this check loads
+the **raw** timestamps exactly as the outlet emitted them
+(`synchronize_clocks=False, dejitter_timestamps=False`) and reports:
+
+- **negative absolute timestamps** — a genuinely bad clock value;
+- **non-monotonic (backward) steps** — consecutive samples whose timestamp goes
+  *backwards*. These are the artifact of stamping each chunk's most-recent
+  sample with `local_clock()` at read time and letting liblsl back-date the
+  rest; when one chunk's back-dated start lands before the previous chunk's end,
+  time runs backward. In a consumer that does **not** dejitter (e.g. Orion), or
+  across machines, those backward steps are what surface as *negative
+  timestamps* / negative inter-sample intervals.
+
+Pass `--fail-on-bad-timestamps` to make the run exit non-zero when the MCC
+stream has negative or non-monotonic raw timestamps (useful as a regression
+gate once the outlet is fixed to emit monotonic, device-anchored timestamps).
+
 Outputs land in `scripts/audio_latency_test/recordings/`:
 `audio_latency_<tag>.xdf`, `latency_<tag>.png`, `latency_<tag>.npz`, and the
 `mcc_<tag>.log` CLI log.

scripts/audio_latency_test/analysis.py

@@ -45,6 +45,83 @@ def load_from_xdf(path, mcc_name, marker_type="Markers", mcc_channel=0):
     return marker_times, mcc_ts, sig
 
 
+# --------------------------------------------------------------------------- #
+# Timestamp health check
+# --------------------------------------------------------------------------- #
+def check_timestamps(path, mcc_name, marker_type="Markers"):
+    """Verify the *raw* (un-synced, un-dejittered) timestamps are sound.
+
+    ``load_from_xdf`` deliberately loads with ``dejitter_timestamps=True``,
+    which refits each stream's timestamps onto a clean monotonic line and so
+    *hides* upstream defects. This routine instead loads the timestamps exactly
+    as the outlet emitted them and checks for the two problems that surface
+    downstream (e.g. in Orion) as "negative timestamps":
+
+    * **negative absolute timestamps** -- a genuinely bad clock value;
+    * **non-monotonic (backward) steps** -- consecutive samples whose timestamp
+      goes *backwards*. These become negative inter-sample intervals (and, in a
+      consumer that derives sample times by accumulating dt or that aligns
+      without dejittering, negative timestamps). They are the expected artifact
+      of stamping each chunk's most-recent sample with ``local_clock()`` at read
+      time and letting liblsl back-date the rest: when one chunk's back-dated
+      start lands before the previous chunk's end, time runs backward.
+
+    Returns ``(ok, results)`` where ``ok`` is False if any stream has negative
+    timestamps, and ``results`` maps stream name -> stats dict. Prints a
+    human-readable summary.
+    """
+    import pyxdf
+
+    streams, _ = pyxdf.load_xdf(
+        path, synchronize_clocks=False, dejitter_timestamps=False)
+
+    results = {}
+    ok = True
+    lines = ["Timestamp health check (raw, as emitted -- no sync/dejitter):"]
+    for s in streams:
+        name = s["info"]["name"][0]
+        ts = np.asarray(s["time_stamps"], dtype=float)
+        if ts.size == 0:
+            continue
+        neg = int(np.count_nonzero(ts < 0))
+        dt = np.diff(ts)
+        back = dt < 0
+        n_back = int(np.count_nonzero(back))
+        max_back_ms = float(-dt[back].min() * 1e3) if n_back else 0.0
+        results[name] = {
+            "n": int(ts.size),
+            "t_min": float(ts.min()), "t_max": float(ts.max()),
+            "n_negative": neg,
+            "n_backward": n_back, "max_backward_ms": max_back_ms,
+            "pct_backward": 100.0 * n_back / max(dt.size, 1),
+        }
+        if neg:
+            status = "FAIL (negative timestamps)"
+            ok = False
+        elif n_back:
+            status = "WARN (non-monotonic)"
+        else:
+            status = "OK"
+        detail = f"  backward_steps={n_back}"
+        if n_back:
+            detail += (f" ({100.0 * n_back / max(dt.size, 1):.2f}% of samples,"
+                       f" max {max_back_ms:.3f} ms back)")
+        lines.append(
+            f"  {name:14} n={ts.size:>8d}  t_min={ts.min():14.4f}"
+            f"  neg={neg}{detail}  -> {status}")
+
+    mcc = results.get(mcc_name)
+    if mcc is not None and mcc["n_negative"] == 0 and mcc["n_backward"] > 0:
+        lines.append(
+            "  NOTE: no negative timestamps on this machine, but the MCC stream "
+            "is non-monotonic. Across machines (or in a consumer that does not "
+            "dejitter, e.g. Orion) those backward steps are what appear as "
+            "negative timestamps. Fix upstream by making the outlet emit "
+            "monotonic, device-anchored timestamps.")
+    print("\n".join(lines))
+    return ok, results
+
+
 # --------------------------------------------------------------------------- #
 # Core: per-marker onset detection
 # --------------------------------------------------------------------------- #

scripts/audio_latency_test/run_test.py

@@ -111,6 +111,11 @@ def parse_args(argv=None):
     p.add_argument("--analyze-xdf", default=None,
                    help="Analyze an existing XDF and exit (no hardware)")
 
+    # Timestamp verification
+    p.add_argument("--fail-on-bad-timestamps", action="store_true",
+                   help="Exit non-zero if the MCC stream has negative or "
+                        "non-monotonic (backward) raw timestamps")
+
     return p.parse_args(argv)
 
 
@@ -156,6 +161,22 @@ def report(args, latencies, marker_times, mcc_ts, mcc_sig, info, tag):
     return stats
 
 
+def verify_timestamps(args, xdf_path):
+    """Run the raw-timestamp health check; return an exit-code contribution.
+
+    Always prints the report. Returns 1 (so callers can OR it into their exit
+    code) when --fail-on-bad-timestamps is set and the MCC stream has negative
+    or non-monotonic timestamps; otherwise 0.
+    """
+    print()
+    ok, results = analysis.check_timestamps(xdf_path, args.mcc_name)
+    if not args.fail_on_bad_timestamps:
+        return 0
+    mcc = results.get(args.mcc_name, {})
+    bad = (not ok) or mcc.get("n_backward", 0) > 0
+    return 1 if bad else 0
+
+
 def _terminate(proc, timeout=5.0):
     if proc is None or proc.poll() is not None:
         return
@@ -185,7 +206,7 @@ def run_analyze_only(args):
     tag = os.path.splitext(os.path.basename(args.analyze_xdf))[0]
     os.makedirs(args.outdir, exist_ok=True)
     report(args, latencies, marker_times, mcc_ts, mcc_sig, info, tag)
-    return 0
+    return verify_timestamps(args, args.analyze_xdf)
 
 
 # --------------------------------------------------------------------------- #
@@ -286,7 +307,7 @@ def run_live(args):
             window=args.window, pre=args.pre,
             thresh_frac=args.thresh_frac, abs_threshold=args.abs_threshold)
         report(args, latencies, marker_times, mcc_ts, mcc_sig, info_recs, tag)
-        return 0
+        return verify_timestamps(args, xdf_path)
 
     finally:
         for closer in (

src/core/CMakeLists.txt

@@ -6,6 +6,18 @@ add_library(mccoutlet_core STATIC
     src/StreamThread.cpp
 )
 
+# Power management: keep the host awake during acquisition. macOS needs the
+# Objective-C++ implementation (NSProcessInfo activity, built with ARC) linked
+# against Foundation; other platforms use the no-op stub.
+if(APPLE)
+    target_sources(mccoutlet_core PRIVATE src/PowerAssertion.mm)
+    set_source_files_properties(src/PowerAssertion.mm
+        PROPERTIES COMPILE_OPTIONS "-fobjc-arc")
+    target_link_libraries(mccoutlet_core PRIVATE "-framework Foundation")
+else()
+    target_sources(mccoutlet_core PRIVATE src/PowerAssertion.cpp)
+endif()
+
 target_include_directories(mccoutlet_core
     PUBLIC
         $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/include>

src/core/include/mccoutlet/Device.hpp

@@ -8,6 +8,7 @@
  */
 
 #include <atomic>
+#include <chrono>
 #include <cstdint>
 #include <functional>
 #include <memory>
@@ -57,6 +58,17 @@ class IDevice {
     virtual bool startAcquisition() = 0;
     virtual void disconnect() = 0;
     virtual bool isConnected() const = 0;
+
+    /**
+     * @brief Request that any in-progress getData*() call return promptly.
+     *
+     * Safe to call from a thread other than the one running getData*().
+     * Causes the blocking acquisition loops to break out so the streaming
+     * thread can observe its shutdown flag and exit before disconnect()
+     * and the subsequent join(). Without this, a wedged device leaves
+     * getData*() spinning and stop() deadlocks on the join.
+     */
+    virtual void requestStop() = 0;
     virtual DeviceInfo getInfo() const = 0;
     virtual DeviceCapabilities getCapabilities() const = 0;
 
@@ -123,6 +135,7 @@ class MCCDevice : public IDevice {
     bool connect() override;
     bool startAcquisition() override;
     void disconnect() override;
+    void requestStop() override;
     bool isConnected() const override;
     DeviceInfo getInfo() const override;
     DeviceCapabilities getCapabilities() const override;
@@ -141,6 +154,10 @@ class MCCDevice : public IDevice {
     std::atomic<bool> disconnecting_{false};
     int overrun_count_ = 0;
 
+    // Coalesce overrun status reports: time of the last one emitted to the
+    // callback. Default-constructed (epoch) means none emitted yet.
+    std::chrono::steady_clock::time_point last_overrun_report_{};
+
     // ulAInScan circular buffer (double precision, as required by uldaq)
     std::vector<double> scan_buffer_;
     int scan_buffer_samples_per_chan_ = 0;

src/core/include/mccoutlet/PowerAssertion.hpp

@@ -0,0 +1,37 @@
+#pragma once
+/**
+ * @file PowerAssertion.hpp
+ * @brief RAII guard that keeps the host awake during acquisition.
+ */
+
+#include <string>
+
+namespace mccoutlet {
+
+/**
+ * @brief Keeps the system awake for as long as the object is alive.
+ *
+ * On macOS, idle system sleep suspends the USB bus, which wedges the DAQ's
+ * in-flight transfers (uldaq then logs "##### error still xfer pending" and
+ * acquisition cannot be recovered). App Nap can likewise throttle the libusb
+ * event thread enough to overrun the device's hardware FIFO. This guard opts
+ * the process out of both via @c -[NSProcessInfo beginActivityWithOptions:reason:]
+ * for the duration of streaming, releasing the activity on destruction.
+ *
+ * On non-Apple platforms it is a no-op (idle sleep does not suspend USB the
+ * same way; add a platform implementation here if that changes).
+ */
+class PowerAssertion {
+public:
+    /// @param reason Human-readable reason shown in power diagnostics.
+    explicit PowerAssertion(const std::string& reason);
+    ~PowerAssertion();
+
+    PowerAssertion(const PowerAssertion&) = delete;
+    PowerAssertion& operator=(const PowerAssertion&) = delete;
+
+private:
+    void* token_ = nullptr;  ///< retained activity token (id<NSObject> on macOS)
+};
+
+} // namespace mccoutlet

src/core/src/Device.cpp

@@ -18,6 +18,11 @@ constexpr int MAX_DEV_COUNT = 100;
 // Larger values tolerate more jitter but use more memory.
 constexpr int SCAN_BUFFER_SECONDS = 10;
 
+// Give up after this many consecutive scan restarts within a single getData*()
+// call. A genuine overrun recovers in one restart; an unrecoverable USB state
+// (e.g. stuck transfers) otherwise spins here forever, hammering ulAInScanStop.
+constexpr int MAX_RESTART_ATTEMPTS = 3;
+
 std::string ulErrorString(UlError err) {
     char msg[ERR_MSG_LEN]{};
     ulGetErrMsg(err, msg);
@@ -413,7 +418,13 @@ bool MCCDevice::restartScan() {
     scans_read_ = 0;
     overrun_count_++;
 
-    if (statusCallback_) {
+    // Coalesce reports to at most one per second so a recovery storm cannot
+    // flood the consumer (e.g. the GUI's queued-event status updates).
+    auto now = std::chrono::steady_clock::now();
+    if (statusCallback_ &&
+        (last_overrun_report_.time_since_epoch().count() == 0 ||
+         now - last_overrun_report_ >= std::chrono::seconds(1))) {
+        last_overrun_report_ = now;
         statusCallback_(
             "Device FIFO overrun detected, scan restarted (overrun #" +
             std::to_string(overrun_count_) + ")", true);
@@ -422,6 +433,12 @@ bool MCCDevice::restartScan() {
     return true;
 }
 
+void MCCDevice::requestStop() {
+    // Break the getData*() polling loops so the streaming thread can exit
+    // before disconnect()/join(). Set from the caller's thread.
+    disconnecting_ = true;
+}
+
 void MCCDevice::disconnect() {
     disconnecting_ = true;
 
@@ -470,13 +487,23 @@ bool MCCDevice::getData(std::vector<float>& buffer, double& timestamp) {
     const int channelCount = config_.high_channel - config_.low_channel + 1;
     const size_t total_buffer_elements = scan_buffer_.size();
 
+    int restart_attempts = 0;
     while (!disconnecting_) {
         ScanStatus status{};
         TransferStatus xfer{};
 
         UlError err = ulAInScanStatus(handle_, &status, &xfer);
         if (err != ERR_NO_ERROR || status != SS_RUNNING) {
             if (disconnecting_) return false;
+            if (++restart_attempts > MAX_RESTART_ATTEMPTS) {
+                if (statusCallback_) {
+                    statusCallback_(
+                        "Acquisition could not be recovered after " +
+                        std::to_string(MAX_RESTART_ATTEMPTS) +
+                        " restart attempts; stopping stream", true);
+                }
+                return false;
+            }
             if (!restartScan()) return false;
             continue;
         }
@@ -514,13 +541,23 @@ bool MCCDevice::getDataInt32(std::vector<int32_t>& buffer, double& timestamp) {
     const int channelCount = config_.high_channel - config_.low_channel + 1;
     const size_t total_buffer_elements = scan_buffer_.size();
 
+    int restart_attempts = 0;
     while (!disconnecting_) {
         ScanStatus status{};
         TransferStatus xfer{};
 
         UlError err = ulAInScanStatus(handle_, &status, &xfer);
         if (err != ERR_NO_ERROR || status != SS_RUNNING) {
             if (disconnecting_) return false;
+            if (++restart_attempts > MAX_RESTART_ATTEMPTS) {
+                if (statusCallback_) {
+                    statusCallback_(
+                        "Acquisition could not be recovered after " +
+                        std::to_string(MAX_RESTART_ATTEMPTS) +
+                        " restart attempts; stopping stream", true);
+                }
+                return false;
+            }
             if (!restartScan()) return false;
             continue;
         }
@@ -560,13 +597,23 @@ bool MCCDevice::getDataInt16(std::vector<int16_t>& buffer, double& timestamp) {
     const int channelCount = config_.high_channel - config_.low_channel + 1;
     const size_t total_buffer_elements = scan_buffer_.size();
 
+    int restart_attempts = 0;
     while (!disconnecting_) {
         ScanStatus status{};
         TransferStatus xfer{};
 
         UlError err = ulAInScanStatus(handle_, &status, &xfer);
         if (err != ERR_NO_ERROR || status != SS_RUNNING) {
             if (disconnecting_) return false;
+            if (++restart_attempts > MAX_RESTART_ATTEMPTS) {
+                if (statusCallback_) {
+                    statusCallback_(
+                        "Acquisition could not be recovered after " +
+                        std::to_string(MAX_RESTART_ATTEMPTS) +
+                        " restart attempts; stopping stream", true);
+                }
+                return false;
+            }
             if (!restartScan()) return false;
             continue;
         }

src/core/src/PowerAssertion.cpp

@@ -0,0 +1,10 @@
+// Non-Apple no-op implementation of PowerAssertion.
+// macOS uses PowerAssertion.mm instead (selected in src/core/CMakeLists.txt).
+#include "mccoutlet/PowerAssertion.hpp"
+
+namespace mccoutlet {
+
+PowerAssertion::PowerAssertion(const std::string& /*reason*/) {}
+PowerAssertion::~PowerAssertion() = default;
+
+} // namespace mccoutlet