From 15b2612d7faacd6bb145bb28406aedf3cf74e680 Mon Sep 17 00:00:00 2001 From: Danny Date: Mon, 2 Nov 2026 16:52:21 -0800 Subject: [PATCH] Hold RFComm for a moment, Better Traces. --- .../system/starport/bugsink_report.py | 26 +++--- openpilot/starpilot/system/starport/sdp.py | 6 +- .../starpilot/system/starport/supervisor.py | 42 ++++++++- .../starport/tests/test_bugsink_report.py | 12 +++ .../system/starport/tests/test_stray_link.py | 87 +++++++++++++++++++ 5 files changed, 157 insertions(+), 16 deletions(-) diff --git a/openpilot/starpilot/system/starport/bugsink_report.py b/openpilot/starpilot/system/starport/bugsink_report.py index c174225fc5..81bf6a89f1 100644 --- a/openpilot/starpilot/system/starport/bugsink_report.py +++ b/openpilot/starpilot/system/starport/bugsink_report.py @@ -27,8 +27,8 @@ MAX_FAILURES = 5 # distinct failures detailed per drive report MAX_BREADCRUMBS = 100 TRACE_QUIET = frozenset({"nav_guidance_sent", "nav_focus", "nav_focus_requested", "nav_status_state", "media_key", "media_audio_state", "media_audio_focus", "media_status_state", "hfp_command", "video_focus"}) -TRACE_SESSION_BYTES = 40 << 10 # per failed session; the oldest events go first -TRACE_TOTAL_BYTES = 160 << 10 # all traces in one report, well under the sender's event cap +TRACE_SESSION_BYTES = 64 << 10 # per failed session; the middle goes first +TRACE_TOTAL_BYTES = 200 << 10 # all traces in one report, well under the sender's event cap EVENT_BUDGET = 400 << 10 # a report is shed down to this before sending; the sender refuses more than 512 KB MAX_ATTENTION = 6 # sessions per report that get a trace and a device slice SERIES_KEYS = ("fps", "encode_ms", "frame_age_p95_ms", "wifi_dbm", "wifi_link", "wifi_noise", "load1", "mem_available_mb", @@ -151,15 +151,19 @@ def quiet_counts(events: list[dict]) -> dict: def bounded_trace(trace: list[dict], limit: int) -> dict: - """The newest events that fit in ``limit`` bytes of JSON, with how many older ones were left out.""" - kept, size = [], 0 - for event in reversed(trace): - size += len(json.dumps(event, default=str)) + 2 - if size > limit: - break - kept.append(event) - kept.reverse() - return {"events": kept, "omitted": len(trace) - len(kept)} + """The first quarter and the newest events that fit in ``limit`` bytes of JSON, with how many in between were left out.""" + sizes = [len(json.dumps(event, default=str)) + 2 for event in trace] + if sum(sizes) <= limit: + return {"events": list(trace), "omitted": 0} + head_end, used = 0, 0 + while head_end < len(trace) and used + sizes[head_end] <= limit // 4: + used += sizes[head_end] + head_end += 1 + tail_start = len(trace) + while tail_start > head_end and used + sizes[tail_start - 1] <= limit: + tail_start -= 1 + used += sizes[tail_start] + return {"events": trace[:head_end] + trace[tail_start:], "omitted": tail_start - head_end, "head": head_end} MAX_OFFERED_MODES = 16 diff --git a/openpilot/starpilot/system/starport/sdp.py b/openpilot/starpilot/system/starport/sdp.py index 7fcbc79150..85a2d66807 100644 --- a/openpilot/starpilot/system/starport/sdp.py +++ b/openpilot/starpilot/system/starport/sdp.py @@ -30,7 +30,9 @@ ERRORS = {1: "invalid SDP version", 2: "invalid service record handle", 3: "inva class SdpError(RuntimeError): - pass + def __init__(self, message: str, raw: bytes = b""): + super().__init__(message) + self.raw = raw # what the car sent, for the log when the record is not usable def build_request(continuation: bytes, service: uuid.UUID = AA_WIRELESS_UUID, transaction_id: int = 0) -> bytes: @@ -107,5 +109,5 @@ def query_channel(sock, timeout: float = 10.0) -> int: raise SdpError("Head unit does not advertise Android Auto Wireless (SDP record missing)") channel = rfcomm_channel(bytes(collected)) if channel is None: - raise SdpError("Android Auto Wireless SDP record has no RFCOMM channel") + raise SdpError("Android Auto Wireless SDP record has no RFCOMM channel", bytes(collected)) return channel diff --git a/openpilot/starpilot/system/starport/supervisor.py b/openpilot/starpilot/system/starport/supervisor.py index 37376dd798..8db49b4e27 100644 --- a/openpilot/starpilot/system/starport/supervisor.py +++ b/openpilot/starpilot/system/starport/supervisor.py @@ -61,6 +61,8 @@ ENCODER_RECOVERIES = 3 # hardware encoder reopens allowed per window befor ENCODER_RECOVERY_WINDOW = 10.0 CAR_LINK_HOLD = 10.0 # a hands-free connection from the car counts as "car present" this long HFP_FRESH = 3.0 # a hands-free link from the car this recent still means "the car is reaching out now" +COMMA_PAGE_WINDOW = 6.0 # a hands-free link this soon after we paged the car is ours; later is the car's +RFCOMM_BUSY_WAITS = (1.0, 2.0) # the kernel can still hold the last session's channel for a moment CAR_FIRST_WAIT = 8.0 # at ignition, give the car this long to open hands-free before the comma pages it NO_STARPORT_WAIT = 8.0 # a paired device must offer wireless projection within this long USB_HANDSHAKE_WAIT = 10.0 # the car connected but sent no AOA START this long -> present as an accessory directly @@ -429,6 +431,8 @@ class Supervisor: self._pairing_attempt: PairingAttempt | None = None self._pairing_seen: tuple = () self._car_seen_at = -CAR_LINK_HOLD + self._paged_at = -COMMA_PAGE_WINDOW # when this comma last asked BlueZ to connect to the car + self._attempt_ended_at: float | None = None self._hfp_link = threading.Event() # set when the chosen car opens the hands-free link self._modern_reply: dict[str, bool] = {} # car address -> answer its v4+ version request like a current phone self._ladder = setup_ladder.SetupLadder(identity_store.CONFIG_PATH.with_name("setup_state.json")) # which handshake to try for each car, and which one last worked @@ -1110,7 +1114,10 @@ class Supervisor: def _hfp_connected(self, address: str) -> None: if address.upper() in (self.config["receiver_address"].upper(), self.config.get("companion_address", "").upper()): - self._car_seen_at = time.monotonic() + now = time.monotonic() + since_page = now - self._paged_at + self.log("hfp_initiator", initiator="comma" if since_page < COMMA_PAGE_WINDOW else "car", since_page_s=round(since_page, 1)) + self._car_seen_at = now self._hfp_link.set() def _release_phone(self) -> None: @@ -1327,6 +1334,7 @@ class Supervisor: car_first = True # the car pages right back once its link drops; let it, rather than crossing it again finally: self._close_sockets() + self._attempt_ended_at = time.monotonic() try: lease.release(restore=False) except Exception as error: @@ -1402,6 +1410,7 @@ class Supervisor: if delay := car_profiles.tuning(config)["connect_delay_s"]: self.log("car_profile_connect_delay", seconds=delay) self._wait(delay) + self._paged_at = time.monotonic() bluez.connect_device(address) self._wait(1.0) @@ -1426,7 +1435,7 @@ class Supervisor: channel_type=strategy.channel_type, patience=strategy.patience, experiments=strategy.experiments) try: self._stage("rfcomm", f"channel {channel}") - rfcomm = self._track(bt_sockets.connect_rfcomm(address, channel)) + rfcomm = self._connect_rfcomm(address, channel) boot = WirelessBootstrap(rfcomm, self._bootstrap_log, device_serial=config["device_name"], version_status=int(config.get("version_status", 0)), experiment="all" if car_profiles.tuning(config)["bootstrap_experiments"] or strategy.experiments else "", @@ -1465,6 +1474,33 @@ class Supervisor: finally: keepalive_stop.set() + def _connect_rfcomm(self, address: str, channel: int): + from openpilot.starpilot.system.starport import bt_sockets + ended = self._attempt_ended_at + since_end = round(time.monotonic() - ended, 1) if ended is not None else None + waited = 0.0 + for tries, delay in enumerate(RFCOMM_BUSY_WAITS, 1): + try: + sock = self._track(bt_sockets.connect_rfcomm(address, channel)) + except OSError as error: + if error.errno != errno.EBUSY: + raise + self.log("rfcomm_busy_retry", channel=channel, wait_s=delay, since_last_attempt_s=since_end) + self._wait(delay) + waited += delay + continue + if tries > 1: + self.log("rfcomm_busy_cleared", channel=channel, tries=tries, waited_s=waited, since_last_attempt_s=since_end) + return sock + try: + sock = self._track(bt_sockets.connect_rfcomm(address, channel)) + except OSError as error: + if error.errno == errno.EBUSY: + self.log("rfcomm_busy_gave_up", channel=channel, waited_s=waited, since_last_attempt_s=since_end) + raise + self.log("rfcomm_busy_cleared", channel=channel, tries=len(RFCOMM_BUSY_WAITS) + 1, waited_s=waited, since_last_attempt_s=since_end) + return sock + def _discover_channel(self, address: str) -> int: from openpilot.starpilot.system.starport import bt_sockets, sdp last_error: Exception | None = None @@ -1478,7 +1514,7 @@ class Supervisor: if _stray_link(error): raise StrayLinkError(f"Bluetooth link to the car is stuck: {error}") from error # retrying cannot clear it last_error = error - self.log("sdp_retry", error=str(error), settle=settle) + self.log("sdp_retry", error=str(error), settle=settle, **({"record_hex": error.raw.hex()[:400]} if getattr(error, "raw", b"") else {})) self._wait(0.35) raise RuntimeError(f"Could not find the car's Android Auto service: {last_error}") diff --git a/openpilot/starpilot/system/starport/tests/test_bugsink_report.py b/openpilot/starpilot/system/starport/tests/test_bugsink_report.py index 396776b379..dd41ff85eb 100644 --- a/openpilot/starpilot/system/starport/tests/test_bugsink_report.py +++ b/openpilot/starpilot/system/starport/tests/test_bugsink_report.py @@ -515,3 +515,15 @@ def test_report_says_how_the_rfcomm_channel_was_found(tmp_path): events.insert(2, (0, 2, "rfcomm_channel", {"channel": 8, "source": "sdp"})) assert br.session_record(write_session(tmp_path, 1, events))["sdp"] == {"channel": 8, "source": "sdp", "retries": 1} assert br.session_record(write_session(tmp_path, 2, STREAMED))["sdp"] == {} + + +def test_trace_keeps_how_the_session_started_as_well_as_how_it_ended(tmp_path): + noisy = [(0, 0, "session_start", {}), (0, 1, "hfp_initiator", {"initiator": "car"})] + noisy += [(i // 60, i % 60, "sdp_retry", {"error": "x" * 250, "n": i}) for i in range(2, 400)] + noisy.append((9, 0, "attempt_failed", {"stage": "discovering", "error": "gone", "kind": "RuntimeError"})) + event = br.drive_event([br.session_record(write_session(tmp_path, 1, noisy))]) + trace = event["contexts"]["starport_trace_session_1"] + names = [e["event"] for e in trace["events"]] + assert names[:2] == ["session_start", "hfp_initiator"] and names[-2:] == ["attempt_failed", "session_stop"] + assert trace["omitted"] > 0 and 0 < trace["head"] < len(names) + assert len(json.dumps(trace)) <= br.TRACE_SESSION_BYTES + 1024 diff --git a/openpilot/starpilot/system/starport/tests/test_stray_link.py b/openpilot/starpilot/system/starport/tests/test_stray_link.py index 462ff78ac3..5ddd77e52d 100644 --- a/openpilot/starpilot/system/starport/tests/test_stray_link.py +++ b/openpilot/starpilot/system/starport/tests/test_stray_link.py @@ -179,3 +179,90 @@ def test_disconnect_stray_link_without_a_handle_only_logs(monkeypatch): phone, commands, events = _phone_with(monkeypatch, {"journalctl": ("", 0)}) assert phone.disconnect_stray_link() is None assert len(commands) == 1 and events == ["stray_link_not_found"] + + +def _rfcomm_attempts(sup, monkeypatch, outcomes): + from openpilot.starpilot.system.starport import bt_sockets + calls = [] + + def connect(*args, **kwargs): + calls.append(args) + outcome = outcomes[min(len(calls), len(outcomes)) - 1] + if isinstance(outcome, int): + raise OSError(outcome, "boom") + return outcome + + monkeypatch.setattr(bt_sockets, "connect_rfcomm", connect) + return calls + + +def test_busy_rfcomm_channel_is_retried_in_the_same_attempt(identity, tmp_path, monkeypatch): + sup, _ = make_supervisor(identity, tmp_path, monkeypatch, lambda *a, **k: socket.socketpair()[0]) + _virtual_time(sup, monkeypatch) + good = socket.socketpair()[0] + calls = _rfcomm_attempts(sup, monkeypatch, [errno.EBUSY, good]) + assert sup._connect_rfcomm(CAR, 6) is good + assert len(calls) == 2 and len(_events(sup, "rfcomm_busy_retry")) == 1 + + +def test_busy_rfcomm_channel_gives_up_after_the_waits(identity, tmp_path, monkeypatch): + from openpilot.starpilot.system.starport import supervisor + sup, _ = make_supervisor(identity, tmp_path, monkeypatch, lambda *a, **k: socket.socketpair()[0]) + _virtual_time(sup, monkeypatch) + calls = _rfcomm_attempts(sup, monkeypatch, [errno.EBUSY]) + with pytest.raises(OSError) as raised: + sup._connect_rfcomm(CAR, 6) + assert raised.value.errno == errno.EBUSY and len(calls) == len(supervisor.RFCOMM_BUSY_WAITS) + 1 + + +def test_other_rfcomm_errors_are_not_retried(identity, tmp_path, monkeypatch): + sup, _ = make_supervisor(identity, tmp_path, monkeypatch, lambda *a, **k: socket.socketpair()[0]) + _virtual_time(sup, monkeypatch) + calls = _rfcomm_attempts(sup, monkeypatch, [errno.ECONNRESET]) + with pytest.raises(OSError): + sup._connect_rfcomm(CAR, 6) + assert len(calls) == 1 and not _events(sup, "rfcomm_busy_retry") + + +def test_a_retried_busy_channel_logs_how_it_ended(identity, tmp_path, monkeypatch): + sup, _ = make_supervisor(identity, tmp_path, monkeypatch, lambda *a, **k: socket.socketpair()[0]) + now = _virtual_time(sup, monkeypatch) + sup._attempt_ended_at = now[0] - 4.0 + _rfcomm_attempts(sup, monkeypatch, [errno.EBUSY, errno.EBUSY, socket.socketpair()[0]]) + sup._connect_rfcomm(CAR, 6) + assert _events(sup, "rfcomm_busy_cleared")[0]["tries"] == 3 + assert _events(sup, "rfcomm_busy_retry")[0]["since_last_attempt_s"] == 4.0 + + +def test_a_channel_that_stays_busy_logs_that_it_gave_up(identity, tmp_path, monkeypatch): + sup, _ = make_supervisor(identity, tmp_path, monkeypatch, lambda *a, **k: socket.socketpair()[0]) + _virtual_time(sup, monkeypatch) + _rfcomm_attempts(sup, monkeypatch, [errno.EBUSY]) + with pytest.raises(OSError): + sup._connect_rfcomm(CAR, 6) + assert _events(sup, "rfcomm_busy_gave_up")[0]["waited_s"] == 3.0 + + +def test_hands_free_logs_who_started_the_link(identity, tmp_path, monkeypatch): + sup, _ = make_supervisor(identity, tmp_path, monkeypatch, lambda *a, **k: socket.socketpair()[0]) + now = _virtual_time(sup, monkeypatch) + sup._paged_at = now[0] - 1.0 + sup._hfp_connected(CAR) + sup._paged_at = now[0] - 60.0 + sup._hfp_connected(CAR) + assert [e["initiator"] for e in _events(sup, "hfp_initiator")] == ["comma", "car"] + + +def test_unusable_sdp_record_is_logged_with_what_the_car_sent(identity, tmp_path, monkeypatch): + from openpilot.starpilot.system.starport import bt_sockets, sdp + sup, _ = make_supervisor(identity, tmp_path, monkeypatch, lambda *a, **k: socket.socketpair()[0]) + _virtual_time(sup, monkeypatch) + monkeypatch.setattr(bt_sockets, "connect_l2cap", lambda *a, **k: socket.socketpair()[0]) + + def query(_): + raise sdp.SdpError("Android Auto Wireless SDP record has no RFCOMM channel", b"\x35\x03\x19\x11\x1f") + + monkeypatch.setattr(sdp, "query_channel", query) + with pytest.raises(RuntimeError): + sup._discover_channel(CAR) + assert _events(sup, "sdp_retry")[0]["record_hex"] == "350319111f"