From d017938b857ac43b52cd60514650fb5bcd200b05 Mon Sep 17 00:00:00 2001 From: Leandro Pineda Date: Tue, 9 Jun 2026 13:24:47 -0300 Subject: [PATCH 1/6] Add supervised background-task helpers to FleetConnector Periodic work placed in `_execution_loop` is already supervised by the framework: `__run_loop` wraps it in try/except, logs the traceback, and retries on the next tick. Work scheduled with a bare `asyncio.create_task` is not: if it raises, the exception is stored on a task nobody awaits (so it is never logged), the task is never restarted, and whatever it fed (e.g. pose / key-values) freezes until the process restarts -- while independent datasources keep working, which masks the failure. Add to FleetConnector: - `_create_supervised_task(name, coro_factory, restart_delay)`: runs a long-lived loop that is logged-with-traceback and restarted if it ever exits or crashes (CancelledError propagates for clean shutdown). - `_spawn_logged_task(coro)`: one-shot tasks whose failures are surfaced via a done-callback instead of dying silently. - Track these tasks and cancel/await them in `__disconnect`. Tests: 3 new cases in tests/test_connector.py. Co-Authored-By: Claude Opus 4.8 (1M context) --- inorbit_connector/connector.py | 91 ++++++++++++++++++++++++++++++++++ tests/test_connector.py | 89 +++++++++++++++++++++++++++++++++ 2 files changed, 180 insertions(+) diff --git a/inorbit_connector/connector.py b/inorbit_connector/connector.py index c0213bd..31f820b 100644 --- a/inorbit_connector/connector.py +++ b/inorbit_connector/connector.py @@ -126,6 +126,10 @@ def __init__(self, config: ConnectorRootConfig, **kwargs) -> None: self.__thread = threading.Thread(target=self.__run_loop) self.__loop: asyncio.AbstractEventLoop | None = None + # Long-lived background tasks scheduled via _create_supervised_task / + # _spawn_logged_task. Cancelled and awaited in __disconnect(). + self.__background_tasks: list[asyncio.Task] = [] + # Additional initalization arguments self.__register_user_scripts = kwargs.get("register_user_scripts", False) self.__default_user_scripts_dir = kwargs.get( @@ -522,6 +526,16 @@ async def __connect(self) -> None: async def __disconnect(self) -> None: """Disconnect external services and disconnect from InOrbit.""" + # Stop supervised background tasks first so their loops stop touching + # robot sessions before we tear those sessions down. Clear the list + # up front so the done-callbacks' self-removal is a no-op. + tasks = self.__background_tasks + self.__background_tasks = [] + for task in tasks: + task.cancel() + if tasks: + await asyncio.gather(*tasks, return_exceptions=True) + # Tear the fleet down while holding the lock so no robot can be added (via # add_robot/update_fleet from another thread) mid-teardown. Free each session # from the pool (not just disconnect it) so a subsequent start() rebuilds and @@ -632,6 +646,83 @@ def _is_session_connected(self, robot_id: str) -> bool: except Exception: return False + def _create_supervised_task( + self, name: str, coro_factory, restart_delay: float = 5.0 + ) -> asyncio.Task: + """Schedule a long-lived background coroutine under supervision. + + Periodic work placed inside ``_execution_loop`` is already supervised by + the framework (``__run_loop`` catches, logs and retries it). Work that + needs a cadence other than ``config.update_freq`` (e.g. a fast pose loop + and a slower key-value loop) must be scheduled here rather than via a + bare ``asyncio.create_task``: a bare task that raises dies silently (its + exception is stored on a task nobody awaits and is never logged), is + never restarted, and freezes that datasource until the process restarts. + + This wraps ``coro_factory`` in a loop that, if the coroutine returns or + raises, logs it (with traceback) and restarts it after ``restart_delay`` + seconds. ``CancelledError`` is propagated so the task stops cleanly on + shutdown. The task is tracked and cancelled in teardown. + + Must be called from the connector's event loop (e.g. from ``_connect``). + + Args: + name: Human-readable name used in log messages. + coro_factory: Zero-arg callable returning a fresh coroutine; called + again on each restart. + restart_delay: Seconds to wait before restarting after exit/crash. + + Returns: + The supervising ``asyncio.Task``. + """ + task = asyncio.create_task(self.__supervise(name, coro_factory, restart_delay)) + task.add_done_callback(self.__log_task_exception) + self.__background_tasks.append(task) + return task + + async def __supervise(self, name, coro_factory, restart_delay) -> None: + """Run ``coro_factory`` forever, logging+restarting it on exit/crash.""" + while True: + try: + await coro_factory() + self._logger.warning( + f"Background task '{name}' exited unexpectedly. " + f"Restarting in {restart_delay}s" + ) + except asyncio.CancelledError: + raise + except Exception: + self._logger.exception( + f"Background task '{name}' crashed. " + f"Restarting in {restart_delay}s" + ) + await asyncio.sleep(restart_delay) + + def _spawn_logged_task(self, coro, name: str | None = None) -> asyncio.Task: + """Schedule a one-shot ``coro`` whose failure is logged, not swallowed. + + Use for fire-and-forget tasks that are not long-lived loops. Unlike a + bare ``asyncio.create_task``, a failure is logged (with traceback) via a + done-callback instead of being stored silently on an un-awaited task. + For long-lived loops use ``_create_supervised_task`` instead. + """ + task = asyncio.create_task(coro) + task.add_done_callback(self.__log_task_exception) + self.__background_tasks.append(task) + return task + + def __log_task_exception(self, task: asyncio.Task) -> None: + """Done-callback: surface a task's exception so it is never silent.""" + try: + self.__background_tasks.remove(task) + except ValueError: + pass + if task.cancelled(): + return + exc = task.exception() + if exc is not None: + self._logger.error(f"Background task failed: {exc!r}", exc_info=exc) + def start(self) -> None: """Start the connector in a new thread. diff --git a/tests/test_connector.py b/tests/test_connector.py index a96d314..19ac3cd 100644 --- a/tests/test_connector.py +++ b/tests/test_connector.py @@ -6,6 +6,7 @@ # SPDX-License-Identifier: MIT # Standard +import asyncio import logging import os import threading @@ -1825,3 +1826,91 @@ class CustomException(Exception): execution_status_details="An error occurred executing custom command", stderr="CustomException", ) + + +# ============================================================================== +# Supervised background tasks +# ============================================================================== + + +class TestSupervisedBackgroundTasks: + """Tests for _create_supervised_task / _spawn_logged_task. + + These guard against the failure mode where a connector schedules periodic + work as a bare asyncio.create_task loop: if it raises, the task dies, the + exception is stored on a task nobody awaits (never logged), nothing restarts + it, and that datasource freezes until the process is restarted. + """ + + @pytest.fixture(autouse=True) + def _concrete_fleet_connector(self): + with _concrete(FleetConnector): + yield + + @pytest.fixture + def connector(self): + return FleetConnector( + ConnectorRootConfig( + api_key="valid_key", + connection_config_url=AnyHttpUrl("https://valid.com/"), + connector_type="valid_connector", + connector_config=DummyConfig(), + fleet=[RobotConfig(robot_id="TestRobot1")], + ) + ) + + @pytest.mark.asyncio + async def test_create_supervised_task_restarts_on_crash(self, connector, caplog): + caplog.set_level(logging.ERROR) + calls = 0 + + async def factory(): + nonlocal calls + calls += 1 + if calls == 1: + raise RuntimeError("boom") + await asyncio.Event().wait() # stay alive once healthy + + task = connector._create_supervised_task("kv-loop", factory, restart_delay=0) + try: + for _ in range(50): + await asyncio.sleep(0.01) + if calls >= 2: + break + finally: + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + assert calls >= 2 # restarted after the crash + assert any("kv-loop" in r.getMessage() for r in caplog.records) + + @pytest.mark.asyncio + async def test_supervised_task_cancelled_on_disconnect(self, connector): + started = asyncio.Event() + + async def factory(): + started.set() + await asyncio.Event().wait() + + task = connector._create_supervised_task("loop", factory, restart_delay=0) + await asyncio.wait_for(started.wait(), timeout=1.0) + + await connector._FleetConnector__disconnect() + + assert task.done() + assert connector._FleetConnector__background_tasks == [] + + @pytest.mark.asyncio + async def test_spawn_logged_task_logs_failure(self, connector, caplog): + caplog.set_level(logging.ERROR) + + async def boom(): + raise ValueError("kaboom") + + task = connector._spawn_logged_task(boom(), name="oneshot") + with pytest.raises(ValueError): + await task + await asyncio.sleep(0) # let the done-callback run + + assert any("kaboom" in r.getMessage() for r in caplog.records) From 52351a9dc7eeaab117bbbda8b85ca58848915055 Mon Sep 17 00:00:00 2001 From: Leandro Pineda Date: Tue, 9 Jun 2026 17:15:33 -0300 Subject: [PATCH 2/6] Require inorbit-edge >=3.1.0 for the MQTT callback-resilience fixes edge-sdk 3.1.0 stops a callback exception from killing the paho network loop thread (guarded callbacks + suppress_exceptions) and surfaces it via enable_logger. Connectors built on this framework get that fix transitively. Co-Authored-By: Claude Opus 4.8 (1M context) --- pyproject.toml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 4430381..e716ba3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -35,7 +35,7 @@ classifiers = [ "License :: OSI Approved :: MIT License", ] dependencies = [ - "inorbit-edge[telemetry]>=3.0,<4.0", + "inorbit-edge[telemetry]>=3.1.0,<4.0", "pydantic>=2.11,<3.0", "pydantic-settings>=2.14,<3.0", "pytz>=2025.1", @@ -45,7 +45,7 @@ dependencies = [ [project.optional-dependencies] video = [ - "inorbit-edge[video]>=3.0,<4.0", + "inorbit-edge[video]>=3.1.0,<4.0", ] dev = [ "bump2version~=1.0", From 15cc6c97c4f7bc1da45c9636121cd316d6f4ace0 Mon Sep 17 00:00:00 2001 From: Leandro Pineda Date: Tue, 9 Jun 2026 17:26:36 -0300 Subject: [PATCH 3/6] Demonstrate _create_supervised_task in the example; TODO metrics + backoff - examples/simple-connector: move odometry polling into a supervised _speed_poll_loop scheduled via _create_supervised_task, showing a loop on a faster cadence than _execution_loop that logs+restarts instead of dying. - __supervise: TODO(metrics) to emit a background_task_errors counter, and TODO(backoff) for capped exponential backoff on repeated crashes. Co-Authored-By: Claude Opus 4.8 (1M context) --- examples/simple-connector/connector.py | 35 ++++++++++++++++++-------- inorbit_connector/connector.py | 9 +++++++ 2 files changed, 33 insertions(+), 11 deletions(-) diff --git a/examples/simple-connector/connector.py b/examples/simple-connector/connector.py index 9aa9fa0..4e5f45f 100644 --- a/examples/simple-connector/connector.py +++ b/examples/simple-connector/connector.py @@ -95,6 +95,28 @@ async def _connect(self) -> None: # Do some magic here... self._logger.info(f"Connected to robot services at API {self.api_version}") + # Schedule a background loop that runs at its own cadence (independent of + # update_freq) under framework supervision: if it raises, the framework + # logs it with a traceback and restarts it, instead of the task dying + # silently and freezing this data source. + self._create_supervised_task("speed_poll_loop", self._speed_poll_loop) + + async def _speed_poll_loop(self) -> None: + """Poll the robot's speeds at ~5Hz and publish odometry. + + Demonstrates ``_create_supervised_task``: a long-lived loop running on a + faster cadence than ``_execution_loop``. + """ + while True: + linear_speed, angular_speed = await asyncio.gather( + get_robot_linear_speed(), + get_robot_angular_speed(), + ) + self.publish_odometry( + linear_speed=linear_speed, angular_speed=angular_speed + ) + await asyncio.sleep(0.2) + @override async def _disconnect(self) -> None: """Disconnect from the robot services.""" @@ -135,17 +157,8 @@ async def _execution_loop(self) -> None: frame_id = "frameIdA" self.publish_pose(x, y, yaw, frame_id) - # A common pattern is to poll REST endpoints for fresh robot data. - # asyncio-compatible libraries are great for this - linear_speed, angular_speed = await asyncio.gather( - get_robot_linear_speed(), - get_robot_angular_speed(), - ) - odometry = { - "linear_speed": linear_speed, - "angular_speed": angular_speed, - } - self.publish_odometry(**odometry) + # Odometry is polled on a faster cadence in _speed_poll_loop, scheduled + # in _connect via _create_supervised_task. self._logger.info("Robot data updated and published") diff --git a/inorbit_connector/connector.py b/inorbit_connector/connector.py index 31f820b..368bef2 100644 --- a/inorbit_connector/connector.py +++ b/inorbit_connector/connector.py @@ -692,10 +692,19 @@ async def __supervise(self, name, coro_factory, restart_delay) -> None: except asyncio.CancelledError: raise except Exception: + # TODO(metrics): increment a `background_task_errors` counter + # (via inorbit_connector.metrics, alongside execution_loop_errors) + # so a crashing/restarting task is visible in monitoring, not + # only in the logs. self._logger.exception( f"Background task '{name}' crashed. " f"Restarting in {restart_delay}s" ) + # TODO(backoff): use a capped exponential backoff keyed per task on + # consecutive failures instead of a fixed restart_delay, so a task + # that fails every iteration (e.g. a persistently malformed upstream + # response) doesn't hammer-restart. Reset it after a run stays up for + # a while. See the MiR connector's circuit breaker for prior art. await asyncio.sleep(restart_delay) def _spawn_logged_task(self, coro, name: str | None = None) -> asyncio.Task: From 738d1fdb183f382eeddea4cf16cad83a9508b9c0 Mon Sep 17 00:00:00 2001 From: Leandro Pineda Date: Tue, 9 Jun 2026 17:38:15 -0300 Subject: [PATCH 4/6] Demonstrate _spawn_logged_task in the example; name logged-task failures - examples/simple-connector: add a one-shot _announce_firmware startup task scheduled via _spawn_logged_task, alongside the supervised speed loop, to show the fire-and-forget-but-logged (not restarted) pattern. - _spawn_logged_task now passes name through to the task so __log_task_exception identifies which task failed (the name arg was previously unused). Co-Authored-By: Claude Opus 4.8 (1M context) --- examples/simple-connector/connector.py | 22 ++++++++++++++++++++++ inorbit_connector/connector.py | 6 ++++-- tests/test_connector.py | 4 +++- 3 files changed, 29 insertions(+), 3 deletions(-) diff --git a/examples/simple-connector/connector.py b/examples/simple-connector/connector.py index 4e5f45f..c25d63d 100644 --- a/examples/simple-connector/connector.py +++ b/examples/simple-connector/connector.py @@ -57,6 +57,12 @@ async def get_robot_angular_speed() -> float: return random.uniform(0.1, 0.9) +async def get_robot_firmware() -> str: + """Simulate a one-time request to the robot's firmware version API.""" + await asyncio.sleep(random.uniform(0.1, 0.3)) + return "fw-2.4.1" + + class ExampleBotConnector(Connector): """The example bot connector. @@ -101,6 +107,11 @@ async def _connect(self) -> None: # silently and freezing this data source. self._create_supervised_task("speed_poll_loop", self._speed_poll_loop) + # One-shot startup work: fire-and-forget, but a failure is logged (not + # lost on an un-awaited task) via _spawn_logged_task. Unlike a + # supervised loop, it runs once and is not restarted. + self._spawn_logged_task(self._announce_firmware(), name="announce_firmware") + async def _speed_poll_loop(self) -> None: """Poll the robot's speeds at ~5Hz and publish odometry. @@ -117,6 +128,17 @@ async def _speed_poll_loop(self) -> None: ) await asyncio.sleep(0.2) + async def _announce_firmware(self) -> None: + """Fetch the firmware version once at startup and publish it. + + Demonstrates ``_spawn_logged_task``: a one-shot fire-and-forget task + whose failure is logged instead of being silently swallowed on an + un-awaited task (but which is not restarted, unlike a supervised loop). + """ + firmware = await get_robot_firmware() + self.publish_key_values(firmware_version=firmware) + self._logger.info(f"Announced firmware version: {firmware}") + @override async def _disconnect(self) -> None: """Disconnect from the robot services.""" diff --git a/inorbit_connector/connector.py b/inorbit_connector/connector.py index 368bef2..0cef8fe 100644 --- a/inorbit_connector/connector.py +++ b/inorbit_connector/connector.py @@ -715,7 +715,7 @@ def _spawn_logged_task(self, coro, name: str | None = None) -> asyncio.Task: done-callback instead of being stored silently on an un-awaited task. For long-lived loops use ``_create_supervised_task`` instead. """ - task = asyncio.create_task(coro) + task = asyncio.create_task(coro, name=name) task.add_done_callback(self.__log_task_exception) self.__background_tasks.append(task) return task @@ -730,7 +730,9 @@ def __log_task_exception(self, task: asyncio.Task) -> None: return exc = task.exception() if exc is not None: - self._logger.error(f"Background task failed: {exc!r}", exc_info=exc) + self._logger.error( + f"Background task '{task.get_name()}' failed: {exc!r}", exc_info=exc + ) def start(self) -> None: """Start the connector in a new thread. diff --git a/tests/test_connector.py b/tests/test_connector.py index 19ac3cd..0057ff0 100644 --- a/tests/test_connector.py +++ b/tests/test_connector.py @@ -1913,4 +1913,6 @@ async def boom(): await task await asyncio.sleep(0) # let the done-callback run - assert any("kaboom" in r.getMessage() for r in caplog.records) + msgs = [r.getMessage() for r in caplog.records] + assert any("kaboom" in m for m in msgs) # the exception is surfaced + assert any("oneshot" in m for m in msgs) # ...tagged with the task name From 34e8e62285383034f08401cd000c9fe001c7b7f8 Mon Sep 17 00:00:00 2001 From: Leandro Pineda Date: Tue, 9 Jun 2026 17:43:17 -0300 Subject: [PATCH 5/6] Align _spawn_logged_task signature with _create_supervised_task Both now lead with a required `name`: _spawn_logged_task(name, coro) to match _create_supervised_task(name, coro_factory, ...). Updates the example and test call sites. Co-Authored-By: Claude Opus 4.8 (1M context) --- examples/simple-connector/connector.py | 2 +- inorbit_connector/connector.py | 10 +++++++++- tests/test_connector.py | 2 +- 3 files changed, 11 insertions(+), 3 deletions(-) diff --git a/examples/simple-connector/connector.py b/examples/simple-connector/connector.py index c25d63d..560606b 100644 --- a/examples/simple-connector/connector.py +++ b/examples/simple-connector/connector.py @@ -110,7 +110,7 @@ async def _connect(self) -> None: # One-shot startup work: fire-and-forget, but a failure is logged (not # lost on an un-awaited task) via _spawn_logged_task. Unlike a # supervised loop, it runs once and is not restarted. - self._spawn_logged_task(self._announce_firmware(), name="announce_firmware") + self._spawn_logged_task("announce_firmware", self._announce_firmware()) async def _speed_poll_loop(self) -> None: """Poll the robot's speeds at ~5Hz and publish odometry. diff --git a/inorbit_connector/connector.py b/inorbit_connector/connector.py index 0cef8fe..1684a68 100644 --- a/inorbit_connector/connector.py +++ b/inorbit_connector/connector.py @@ -707,13 +707,21 @@ async def __supervise(self, name, coro_factory, restart_delay) -> None: # a while. See the MiR connector's circuit breaker for prior art. await asyncio.sleep(restart_delay) - def _spawn_logged_task(self, coro, name: str | None = None) -> asyncio.Task: + def _spawn_logged_task(self, name: str, coro) -> asyncio.Task: """Schedule a one-shot ``coro`` whose failure is logged, not swallowed. Use for fire-and-forget tasks that are not long-lived loops. Unlike a bare ``asyncio.create_task``, a failure is logged (with traceback) via a done-callback instead of being stored silently on an un-awaited task. For long-lived loops use ``_create_supervised_task`` instead. + + Args: + name: Human-readable name, used in the failure log (mirrors + ``_create_supervised_task``'s leading ``name`` argument). + coro: The coroutine to run once. + + Returns: + The scheduled ``asyncio.Task``. """ task = asyncio.create_task(coro, name=name) task.add_done_callback(self.__log_task_exception) diff --git a/tests/test_connector.py b/tests/test_connector.py index 0057ff0..5061f36 100644 --- a/tests/test_connector.py +++ b/tests/test_connector.py @@ -1908,7 +1908,7 @@ async def test_spawn_logged_task_logs_failure(self, connector, caplog): async def boom(): raise ValueError("kaboom") - task = connector._spawn_logged_task(boom(), name="oneshot") + task = connector._spawn_logged_task("oneshot", boom()) with pytest.raises(ValueError): await task await asyncio.sleep(0) # let the done-callback run From 3a83b724c582f147583cd74792a21271ae86db62 Mon Sep 17 00:00:00 2001 From: Leandro Pineda Date: Tue, 9 Jun 2026 22:11:20 -0300 Subject: [PATCH 6/6] Documentation updated --- docs/contents/specification/connector.md | 16 ++++++++++++++-- docs/contents/usage/fleet.md | 24 ++++++++++++++++++++++++ docs/contents/usage/single-robot.md | 22 ++++++++++++++++++++++ 3 files changed, 60 insertions(+), 2 deletions(-) diff --git a/docs/contents/specification/connector.md b/docs/contents/specification/connector.md index 6c9140d..1975686 100644 --- a/docs/contents/specification/connector.md +++ b/docs/contents/specification/connector.md @@ -31,7 +31,7 @@ Typical responsibilities: - Connect to your fleet manager / backend services. - Optionally fetch the fleet membership and call `update_fleet()` **before** sessions are created. -- Start background polling tasks that keep fresh state for all robots. +- Start background polling tasks that keep fresh state for all robots. Schedule them with [`_create_supervised_task()`](#spec-connector-fleetconnector-background-tasks) rather than a bare `asyncio.create_task`, so a crash is logged and the loop restarts instead of dying silently. ### `_execution_loop()` @@ -50,7 +50,19 @@ Typical responsibilities: **Override.** -Called once during shutdown, after Edge SDK sessions are disconnected. Use this to stop polling tasks, close sockets, and release resources. +Called once during shutdown, after Edge SDK sessions are disconnected. Use this to stop polling tasks, close sockets, and release resources. (Tasks scheduled via [`_create_supervised_task()` / `_spawn_logged_task()`](#spec-connector-fleetconnector-background-tasks) are cancelled automatically — you only need to clean up tasks you started with a bare `asyncio.create_task`, which is discouraged.) + + +### `_create_supervised_task(name, coro_factory, restart_delay=5.0)` / `_spawn_logged_task(name, coro)` + +**Callable (advanced).** + +Schedule background work that runs alongside `_execution_loop` — for example a fast pose loop and a slower key-value loop, each on its own cadence. Prefer these over a bare `asyncio.create_task`: a fire-and-forget task that raises has its exception stored on a task nobody awaits, so it is never logged, never restarted, and the datasource it fed silently freezes until the process restarts (while unrelated datasources keep working, masking the failure). + +- **`_create_supervised_task(name, coro_factory, restart_delay=5.0)`** — for long-lived loops. Runs `coro_factory()` and, if it ever returns or raises, logs it (with a traceback, tagged with `name`) and restarts it after `restart_delay` seconds. `coro_factory` is a zero-argument callable returning a fresh coroutine (it is re-called on each restart). `asyncio.CancelledError` is propagated so the task stops cleanly on shutdown. +- **`_spawn_logged_task(name, coro)`** — for one-shot work. Runs `coro` once; if it raises, the failure is logged (with a traceback, tagged with `name`) via a done-callback instead of being silently swallowed. It is **not** restarted. + +Call these from `_connect()` (they require the connector's event loop to be running). Tasks scheduled this way are tracked by the framework and cancelled automatically during shutdown, so they do not need to be stopped in `_disconnect()`. ### `_inorbit_robot_command_handler(robot_id, command_name, args, options)` diff --git a/docs/contents/usage/fleet.md b/docs/contents/usage/fleet.md index 41cb0e3..4e0113e 100644 --- a/docs/contents/usage/fleet.md +++ b/docs/contents/usage/fleet.md @@ -114,6 +114,30 @@ All publishing methods require a `robot_id` parameter. See the [Publishing Guide - `publish_robot_system_stats(robot_id, **kwargs)`: Defer publishing of system stats for a specific robot; defaults are published if not called - `publish_robot_map(robot_id, frame_id, is_update=False)`: Publish map for a specific robot +## Background tasks + +For work that needs a different cadence than `_execution_loop` (e.g. a fast per-robot poll), schedule **supervised** background loops from `_connect()` instead of bare `asyncio.create_task` calls. The framework logs and restarts a loop if it crashes — instead of the task dying silently and freezing that data source — and cancels it automatically on shutdown: + +```python +@override +async def _connect(self) -> None: + self._fleet_api = MyFleetApi(...) + # One supervised loop per robot. coro_factory is zero-arg, so bind the + # robot id with a default-arg lambda: + for robot_id in self.robot_ids: + self._create_supervised_task( + f"poll:{robot_id}", lambda rid=robot_id: self._poll_robot(rid) + ) + +async def _poll_robot(self, robot_id: str) -> None: + while True: + speed = await self._fleet_api.get_speed(robot_id) + self.publish_robot_odometry(robot_id, linear_speed=speed) + await asyncio.sleep(0.2) +``` + +For one-shot work, use `_spawn_logged_task(name, coro)` — its failure is logged (not silently swallowed), but it is not restarted. See the [Connector API spec](../specification/connector#spec-connector-fleetconnector-background-tasks). + ## Advanced Methods ### `_get_robot_session()` diff --git a/docs/contents/usage/single-robot.md b/docs/contents/usage/single-robot.md index f028b6c..e6cceb7 100644 --- a/docs/contents/usage/single-robot.md +++ b/docs/contents/usage/single-robot.md @@ -61,6 +61,28 @@ async def _execution_loop(self) -> None: The loop runs at the frequency specified by `config.update_freq` (default: 1.0 Hz). +### Background tasks + +For work that needs a different cadence than `_execution_loop` (e.g. a fast sensor poll), schedule a **supervised** background loop from `_connect()` instead of a bare `asyncio.create_task`. The framework logs and restarts it if it crashes — instead of the task dying silently and freezing that data source — and cancels it automatically on shutdown: + +```python +@override +async def _connect(self) -> None: + self._robot_api = MyRobotApi(...) + # Long-lived loop on its own cadence, supervised (logged + restarted on crash): + self._create_supervised_task("speed_poll", self._poll_speeds) + +async def _poll_speeds(self) -> None: + while True: + speed = await self._robot_api.get_speed() + self.publish_odometry(linear_speed=speed) + await asyncio.sleep(0.2) +``` + +For one-shot startup work, use `_spawn_logged_task("announce_firmware", self._announce_firmware())` — its failure is logged (not silently swallowed), but it is not restarted. + +See the [Connector API spec](../specification/connector#spec-connector-fleetconnector-background-tasks) and the [simple-connector example](https://github.com/inorbit-ai/inorbit-connector-python/blob/main/examples/simple-connector/connector.py). + ### `_disconnect()` Clean up resources and disconnect from external services. This is called when the connector stops.