Merge pull request #33 from rticommunity/dev

Harden exception/cleanup paths and fix QoS wildcard shadowing

Author: rtidgreenberg

.github/copilot-instructions.md

@@ -5,6 +5,16 @@
 - **Always** use the pre-existing venv: `source python/.venv/bin/activate`
 - The venv has `rti.connext 7.6.0` on Python 3.8. Never use the system Python (has 7.3.1).
 - Run tests from inside the venv: `cd python && source .venv/bin/activate && python -m pytest tests/ -x -q`
+- Before running tests or any DDS code, ensure RTI runtime variables are exported in the same shell session:
+	- `export NDDSHOME=/path/to/rti_connext_dds-7.6.0`
+	- `export PATH="$NDDSHOME/bin:$PATH"`
+	- `export LD_LIBRARY_PATH="$NDDSHOME/lib/${RTI_ARCH:-x64Linux4gcc7.3.0}:$LD_LIBRARY_PATH"`
+	- `export RTI_LICENSE_FILE="${RTI_LICENSE_FILE:-$NDDSHOME/rti_license.dat}"`
+- If `NDDSHOME`/license is missing, `dds.DomainParticipant` creation can fail and tests will error before execution.
+- Before running any test suite, verify the wildcard USTM QoS assignment is commented out in `qos/umaa_qos_lib.xml`:
+	- `<!-- <datawriter_qos topic_filter="*" base_name="USTMQoS" /> -->`
+	- `<!-- <datareader_qos topic_filter="*" base_name="USTMQoS" /> -->`
+- Rationale: topic filters are first-match-wins; uncommenting the wildcard USTM lines shadows all specific QoS mappings and breaks QoS profile tests.
 
 ## Project Layout
 

python/docs/exception-cleanup-findings-2026-05-18.md

@@ -0,0 +1,120 @@
+# Exception and Cleanup Findings (2026-05-18)
+
+This document captures a focused review of exception handling and shutdown/cleanup behavior in the Python SDK runtime lifecycle.
+
+## Scope
+
+- Command provider session lifecycle
+- Command consumer session lifecycle
+- Context startup/shutdown orchestration
+- Hook exception isolation
+
+## Findings Already Handled
+
+### 1. Consumer hook exception isolation is robust
+
+- `on_status`, `on_ack`, `on_exec_status`, and `on_terminal` hook exceptions are caught and logged.
+- Consumer session cleanup still completes even if `on_terminal` raises.
+- Existing tests cover this behavior (`test_consumer_hook_isolation.py`).
+
+### 2. Shutdown is idempotent
+
+- `DDSContext.shutdown()` uses `_is_shutdown` guard and returns on repeated calls.
+
+### 3. Shutdown order is correct for dispatcher safety
+
+- Dispatcher is stopped before task cancellation and DDS entity teardown.
+- Service-level `close()` is used for logical cleanup only.
+
+### 4. Context-level service close failures are contained
+
+- `DDSContext.shutdown()` catches/logs service `close()` exceptions and continues teardown.
+
+## Open Findings (Not Yet Fully Handled)
+
+### 1. Provider `on_terminal` exceptions can skip provider session cleanup
+
+In `CommandProviderSession.run()` finalization, `await self._provider.on_terminal(self)` executes before:
+
+- provider instance disposal
+- active-session map removal
+
+If `on_terminal()` raises, disposal and map cleanup may be skipped for that session.
+
+Risk:
+
+- lingering `_active_sessions` entries
+- missed instance disposal
+- shutdown path inconsistencies under hook failure
+
+### 2. `run_until_shutdown()` can double-start already-started services
+
+`DDSContext.run_until_shutdown()` currently creates a new task for every service exposing `_run()` without checking whether a prior `start()` already created a live task.
+
+Risk:
+
+- duplicate reader loops for services that were manually started
+- hard-to-debug duplicated processing
+
+### 3. Provider `close()` can abort early on non-cancel exception from `_run` task await
+
+`CommandProvider.close()` cancels `_task` and only handles `asyncio.CancelledError` when awaiting it.
+If awaiting `_task` raises a different exception, active-session fail/cleanup logic below may not execute in that `close()` call.
+
+Risk:
+
+- incomplete fail-on-shutdown behavior for active sessions
+- reduced cleanup resilience after reader-loop failure
+
+## Suggested Fixes
+
+### A. Harden provider session finalization
+
+Wrap provider `on_terminal` in `try/except` in the `finally` block, and always run disposal and `_active_sessions.pop(...)` afterward.
+
+Suggested shape:
+
+1. `try: await on_terminal(...)`
+2. `except Exception: log`
+3. always dispose instances
+4. always remove session from active map
+
+### B. Prevent double-start in `run_until_shutdown()`
+
+Before creating a task for `_run()`, check whether `_task` exists and is still running.
+
+Suggested guard:
+
+- start only when `_task is None` or `_task.done()`
+
+### C. Make provider `close()` resilient to non-cancel task failures
+
+When awaiting canceled `_task`, catch generic exceptions (log and continue) so active sessions are still failed and awaited.
+
+## Test Gaps to Add
+
+1. Provider hook isolation test:
+
+- Provider subclass whose `on_terminal()` raises.
+- Assert session still disposes and is removed from `_active_sessions`.
+
+2. Lifecycle double-start guard test:
+
+- Call `service.start()` and then `ctx.run_until_shutdown()`.
+- Assert only one `_run()` task loop executes.
+
+3. Provider close resilience test:
+
+- Force `_run()` task to fail with non-cancel exception.
+- Assert `close()` still proceeds to fail/await active sessions.
+
+## Priority
+
+- High: provider `on_terminal` finalization hardening
+- High: double-start guard in `run_until_shutdown()`
+- Medium: provider `close()` robustness for non-cancel task exceptions
+
+## Notes
+
+- Findings are based on code-path verification in runtime implementation, not architecture docs alone.
+- Consumer exception isolation is in better shape than provider finalization paths.

python/docs/index.rst

@@ -53,6 +53,7 @@ Key facts:
    :caption: Project
 
    changelog
+   exception-cleanup-findings-2026-05-18
 
 
 Indices

python/rtiumaapy/command_provider.py

@@ -254,6 +254,11 @@ async def close(self) -> None:
                 await self._task
             except asyncio.CancelledError:
                 pass
+            except Exception:
+                _logger.exception(
+                    "Provider %s: _run task failed during close",
+                    self.service_name,
+                )
 
         # Fail all active sessions
         tasks = []

python/rtiumaapy/command_provider_session.py

@@ -329,6 +329,12 @@ async def run(self) -> None:
                 await self._provider.on_failed(self, e)
 
         finally:
-            await self._provider.on_terminal(self)
+            try:
+                await self._provider.on_terminal(self)
+            except Exception:
+                _logger.exception(
+                    "Session %s: provider on_terminal hook error",
+                    self._session_id,
+                )
             self._dispose_provider_instances()
             self._provider._active_sessions.pop(self._session_id, None)

python/rtiumaapy/dds_context.py

@@ -302,7 +302,9 @@ async def run_until_shutdown(self) -> None:
         # Start _run() for every registered service
         for service in self._registry.values():
             if hasattr(service, "_run"):
-                service._task = asyncio.create_task(service._run())
+                task: Optional[asyncio.Task] = getattr(service, "_task", None)
+                if task is None or task.done():
+                    service._task = asyncio.create_task(service._run())
 
         # Wait for shutdown signal
         stop = asyncio.Event()

python/tests/test_base_service.py

@@ -43,6 +43,32 @@ async def close(self) -> None:
         self.closed = True
 
 
+class CountingRunService(BaseService):
+    """Service that records how many concurrent _run entries occurred."""
+
+    def __init__(self, ctx: DDSContext, service_name: str) -> None:
+        super().__init__(ctx, service_name)
+        self.closed = False
+        self.run_entries = 0
+        self.duplicate_started = asyncio.Event()
+
+    async def _run(self) -> None:
+        self.run_entries += 1
+        if self.run_entries > 1:
+            self.duplicate_started.set()
+        try:
+            await asyncio.Event().wait()
+        except asyncio.CancelledError:
+            pass
+
+    def start(self) -> None:
+        if not hasattr(self, "_task") or self._task is None or self._task.done():
+            self._task = asyncio.ensure_future(self._run())
+
+    async def close(self) -> None:
+        self.closed = True
+
+
 # ═══════════════════════════════════════════════════════════════════════════
 # Auto-registration
 # ═══════════════════════════════════════════════════════════════════════════
@@ -145,3 +171,28 @@ async def send_signal():
         await ctx.run_until_shutdown()
         assert svc.ran
         assert svc.closed
+
+    @pytest.mark.asyncio
+    async def test_run_until_shutdown_does_not_double_start(
+        self, qos_file: str,
+    ):
+        """If start() already ran, run_until_shutdown() must not spawn another _run task."""
+        import os
+        import signal
+
+        ctx = DDSContext(domain_id=DEFAULT_DOMAIN_ID, qos_file=qos_file)
+        svc = CountingRunService(ctx, "RunnerOnce")
+        svc.start()
+
+        loop = asyncio.get_running_loop()
+
+        async def send_signal():
+            await asyncio.sleep(0.1)
+            loop.call_soon(lambda: os.kill(os.getpid(), signal.SIGINT))
+
+        asyncio.create_task(send_signal())
+        await ctx.run_until_shutdown()
+
+        assert svc.run_entries == 1
+        assert not svc.duplicate_started.is_set()
+        assert svc.closed

python/tests/test_command_shutdown.py

@@ -65,6 +65,28 @@ async def on_executing(self, session):
         await asyncio.sleep(30)
 
 
+class TerminalFailProvider(CommandProvider):
+    def __init__(self, ctx, source_id):
+        super().__init__(
+            ctx, "TerminalFailProvider",
+            command_type=AnchorCommandType,
+            ack_type=AnchorCommandAckReportType,
+            status_type=AnchorCommandStatusType,
+            command_topic=AnchorCommandTypeTopic,
+            ack_topic=AnchorCommandAckReportTypeTopic,
+            status_topic=AnchorCommandStatusTypeTopic,
+            source_id=source_id,
+        )
+        self.terminal_calls = 0
+
+    async def on_executing(self, session):
+        return
+
+    async def on_terminal(self, session):
+        self.terminal_calls += 1
+        raise RuntimeError("terminal hook boom")
+
+
 class ShutdownConsumer(CommandConsumer):
     def __init__(self, ctx, source_id, destination_id):
         super().__init__(
@@ -148,6 +170,56 @@ async def test_close_no_active_sessions(self, dds_context: DDSContext):
         # Should not raise
         await provider.close()
 
+    @pytest.mark.asyncio
+    async def test_provider_on_terminal_error_still_cleans_session(
+        self, dds_context: DDSContext,
+    ):
+        """Provider on_terminal raise should not skip session disposal/pop."""
+        provider_id = _make_id()
+        consumer_id = _make_id()
+
+        provider = TerminalFailProvider(dds_context, source_id=provider_id)
+        consumer = ShutdownConsumer(
+            dds_context, source_id=consumer_id,
+            destination_id=provider_id)
+
+        provider.start()
+        consumer.start()
+
+        await wait_for_match(consumer._command_writer)
+        await wait_for_match(provider._ack_writer)
+        await wait_for_match(provider._status_writer)
+
+        await consumer.send(AnchorCommandType())
+
+        await asyncio.wait_for(
+            consumer.terminal_event.wait(), timeout=10.0)
+
+        # Give provider session finally block a moment to complete.
+        await asyncio.sleep(0.2)
+
+        assert provider.terminal_calls == 1
+        assert provider._active_sessions == {}
+
+    @pytest.mark.asyncio
+    async def test_close_continues_if_run_task_raises_non_cancel(
+        self, dds_context: DDSContext,
+    ):
+        """Provider close should continue cleanup if canceled _run raises."""
+        provider_id = _make_id()
+        provider = SlowProvider(dds_context, source_id=provider_id)
+
+        async def bad_run():
+            try:
+                await asyncio.Event().wait()
+            except asyncio.CancelledError as exc:
+                raise RuntimeError("forced _run failure") from exc
+
+        provider._task = asyncio.create_task(bad_run())
+
+        # Should not raise; close should tolerate non-cancel task failure.
+        await provider.close()
+
 
 class TestConsumerShutdown:
     """Consumer close() during active session (C39)."""

python/tests/test_dds_context.py

@@ -8,6 +8,7 @@
 from rtiumaapy.dds_context import DDSContext
 from tests.conftest import DEFAULT_DOMAIN_ID, SimpleReport
 from rtiumaapy.datamodel.AccelerationReportType import (
+    UMAA_SA_AccelerationStatus_AccelerationReportType as AccelerationReportType,
     UMAA_SA_AccelerationStatus_AccelerationReportTypeTopic as AccelerationReportTypeTopic,
 )
 from rtiumaapy.datamodel.BatteryReportType import (
@@ -17,10 +18,16 @@
     UMAA_SEM_GPSStatus_GPSReportTypeTopic as GPSReportTypeTopic,
 )
 from rtiumaapy.datamodel.AnchorCommandType import (
+    UMAA_EO_AnchorControl_AnchorCommandType as AnchorCommandType,
     UMAA_EO_AnchorControl_AnchorCommandTypeTopic as AnchorCommandTypeTopic,
 )
 
 
+def _reliability_kind_name(kind) -> str:
+    """Normalize RTI reliability kind wrappers/enums to a stable string."""
+    return str(kind).split(".")[-1]
+
+
 # ═══════════════════════════════════════════════════════════════════════════
 # Singleton behaviour
 # ═══════════════════════════════════════════════════════════════════════════
@@ -123,16 +130,22 @@ async def test_creates_writer(self, dds_context: DDSContext):
     @pytest.mark.asyncio
     async def test_report_qos_best_effort(self, dds_context: DDSContext):
         """*ReportType topics should get TelemetryQoS → BEST_EFFORT."""
-        writer = dds_context.create_writer(SimpleReport, AccelerationReportTypeTopic)
+        writer = dds_context.create_writer(
+            AccelerationReportType,
+            AccelerationReportTypeTopic,
+        )
         qos = writer.qos
-        assert qos.reliability.kind == dds.ReliabilityKind.BEST_EFFORT
+        assert _reliability_kind_name(qos.reliability.kind) == "BEST_EFFORT"
 
     @pytest.mark.asyncio
     async def test_command_qos_reliable(self, dds_context: DDSContext):
         """*CommandType topics should get CommandQoS → RELIABLE."""
-        writer = dds_context.create_writer(SimpleReport, AnchorCommandTypeTopic)
+        writer = dds_context.create_writer(
+            AnchorCommandType,
+            AnchorCommandTypeTopic,
+        )
         qos = writer.qos
-        assert qos.reliability.kind == dds.ReliabilityKind.RELIABLE
+        assert _reliability_kind_name(qos.reliability.kind) == "RELIABLE"
 
 
 class TestCreateReader:
@@ -143,9 +156,12 @@ async def test_creates_reader(self, dds_context: DDSContext):
 
     @pytest.mark.asyncio
     async def test_report_qos_best_effort(self, dds_context: DDSContext):
-        reader = dds_context.create_reader(SimpleReport, AccelerationReportTypeTopic)
+        reader = dds_context.create_reader(
+            AccelerationReportType,
+            AccelerationReportTypeTopic,
+        )
         qos = reader.qos
-        assert qos.reliability.kind == dds.ReliabilityKind.BEST_EFFORT
+        assert _reliability_kind_name(qos.reliability.kind) == "BEST_EFFORT"
 
 
 class TestCreateFilteredReader: