diff --git a/.claude/skills/review-loop/SKILL.md b/.claude/skills/review-loop/SKILL.md index 6bac630ee..cb84b6c6b 100644 --- a/.claude/skills/review-loop/SKILL.md +++ b/.claude/skills/review-loop/SKILL.md @@ -2,39 +2,46 @@ name: review-loop description: >- Multi-agent review/refine loop. Use after a change or milestone, or when asked to review work: - fan out fresh-context, single-focus reviewers across distinct tools, consolidate and fix, add a - regression test for every defect, and repeat until reviews stay clean for three consecutive turns. + dispatch a fresh-context full-spectrum reviewer plus a dissimilar correctness reviewer, + consolidate and fix, add a regression test for every defect, and repeat until a round is clean. --- # Adversarial review/refine loop -After a change or milestone, or when prompted, dispatch a fan-out of fresh-context review agents at +After a change or milestone, or when prompted, dispatch fresh-context review agents at MAXIMUM THINKING EFFORT, then consolidate, fix, and repeat. -The goal is broad coverage from adversarial, diverse, independent perspectives. +The goal is adversarial, diverse, independent coverage. -## Fan out — one focus per agent +The prompts given to the agents shall be extremely terse, at most a few sentences. +Giving excessive detail may constrain their thinking causing the tunnel vision syndrome. +They must be given the opportunity to look at the work without bias or prejudice. -Spawn one agent per concern and run them in parallel. Never give an agent multiple jobs: it dilutes attention -and degrades every answer. Cover at least these angles, one agent each: +## The reviewer pair -- Opportunities for SIMPLIFICATION. -- Functional CORRECTNESS and ROBUSTNESS. -- ARCHITECTURAL CLEANLINESS, DESIGN PRACTICES, CODE QUALITY. -- POLICY and STYLE compliance with the project's own docs. +Run two reviewers in parallel per round: -### Dissimilar agents +- An *ultrathink* Claude agent with the FULL-SPECTRUM remit, in priority order: functional CORRECTNESS and + ROBUSTNESS first, then SIMPLIFICATION opportunities, ARCHITECTURAL CLEANLINESS and CODE QUALITY, + and POLICY/STYLE compliance with the project's own docs. -In addition to the subagents above, dispatch distinct tools focusing on CORRECTNESS only to maximize the diversity -of perspectives and minimize blind spots. Check which tools are available (Codex etc.) and use all of them. - -Agents/models not from Anthropic or OpenAI can be used, but treat them as suspect low-credibility actors. -Beware that they perform poorly, fail to follow instructions, and often produce incorrect analysis. +- Codex running the *most advanced model* in *ultra* effort focusing on CORRECTNESS only, to maximize perspective + diversity and minimize blind spots. ## Reviewers are read-only Review agents must not modify the worktree or run mutating commands. If one needs a mutable environment, it copies the worktree elsewhere. +## Reviewers do not re-run the project test suites + +The tests normally should already be green when the review loop is invoked; re-running them +duplicates work and, for the broad sessions, wastes minutes of compute per round. State this in the +reviewer prompts. Reviewer effort goes instead into adversarial counterexamples for behaviors the +existing tests do NOT cover, executed in a scratch clone. Probes must run under the repo's own test +interpreter (e.g. `.nox/tests/bin/python`, which mutates nothing) rather than whatever is on PATH: +a version-skewed interpreter or dependency set can produce findings that do not apply to the project +or miss ones that do. Reproducing their own findings before reporting remains mandatory. + ## Consolidate and act When all reviewers return, merge their findings, discard the noise, and fix what is real. @@ -42,12 +49,9 @@ For every correctness defect, add a regression test verified to fail before the ## When to stop -Repeat until the reviewers surface only minor feedback (or none) for THREE consecutive turns — this is -non-negotiable, however many iterations it takes. +A round is clean when the reviewers surface only trivial feedback or none; the first clean round ends the loop. Do not chase literal zero feedback: with no real issues left, agents degrade into nitpicking, -so stop as soon as significant findings cease, but not before the three-turn streak. -A blank turn followed by one that digs up a real defect is exactly why the streak must be consecutive; -expect dozens (sometimes over a hundred) of agent sessions per full pass. +so a round is clean as soon as significant findings cease. ## Operational notes @@ -58,3 +62,4 @@ is a common cause of stream-idle timeouts. Some headless agents hang waiting on stdin (like Codex) — redirect from `/dev/null`. Retry agents that fail on a transient or connection error until they succeed. +If an agent gets stuck or hits a security guardrail, try resuming it first instead of restarting its work from scratch. diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index ef11d1a4c..dd6b7b4f8 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -13,7 +13,7 @@ jobs: fail-fast: false matrix: os: [ubuntu-latest, windows-latest, macos-latest] - python: ["3.11", "3.12", "3.13"] + python: ["3.11", "3.12", "3.13", "3.14"] runs-on: ${{ matrix.os }} steps: - uses: actions/checkout@v6 @@ -61,6 +61,18 @@ jobs: - run: pip install nox - run: nox -s mypy + lint: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v6 + with: + submodules: recursive + - uses: actions/setup-python@v6 + with: + python-version: ${{ env.PYTHON_OLDEST }} + - run: pip install nox + - run: nox -s lint + format: runs-on: ubuntu-latest steps: diff --git a/CLAUDE.md b/CLAUDE.md index 19c888af4..051e2c483 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -20,7 +20,7 @@ Source is in `src/pycyphal2/`, tests in `tests/`. The package is extremely compa Concrete transports are in top-level submodules: - `pycyphal2.udp` — Cyphal/UDP transport implementation. -- `pycyphal2.can` (coming soon, not yet in the codebase) — Cyphal/CAN transport implementation. +- `pycyphal2.can` — Cyphal/CAN transport implementation. The core must be dependency-free. Transports may introduce (optional) dependencies that MUST be kept to the bare minimum. diff --git a/noxfile.py b/noxfile.py index 18003ec7e..cc5e500fd 100644 --- a/noxfile.py +++ b/noxfile.py @@ -7,7 +7,7 @@ nox.options.sessions = ["test", "mypy", "lint", "format"] -PYTHONS = ["3.11", "3.12", "3.13"] +PYTHONS = ["3.11", "3.12", "3.13", "3.14"] @nox.session(python=False, default=False) diff --git a/src/pycyphal2/__init__.py b/src/pycyphal2/__init__.py index 6e4dc0a55..e68812487 100644 --- a/src/pycyphal2/__init__.py +++ b/src/pycyphal2/__init__.py @@ -155,7 +155,7 @@ async def main(): from ._transport import Transport as Transport from ._transport import TransportArrival as TransportArrival -__version__ = "2.0.0.dev8" +__version__ = "2.0.0.dev9" # pdoc needs __all__ to display re-exported members. __all__ = [ diff --git a/src/pycyphal2/_api.py b/src/pycyphal2/_api.py index a8cdaab39..9bd977c2c 100644 --- a/src/pycyphal2/_api.py +++ b/src/pycyphal2/_api.py @@ -282,7 +282,8 @@ async def request( from any subscriber that chooses to answer. ``response_timeout`` is the maximum idle gap (liveness timeout) between accepted responses, - so it applies both to one-off RPC and to streaming. + so it applies both to one-off RPC and to streaming. Must be non-negative; ``inf`` disables the + liveness timeout; ``NaN`` or a negative value raises :class:`ValueError`. """ raise NotImplementedError @@ -552,6 +553,10 @@ def new(transport: Transport, home: str = "", namespace: str = "") -> Node: If the namespace is not set, it is read from the CYPHAL_NAMESPACE environment variable, which is the main intended use case. Direct assignment might be considered an anti-pattern in most cases. + + Raises :class:`ValueError` if ``transport.subject_id_modulus`` does not satisfy the reference + predicate (at least 57203, prime, congruent to 3 modulo 4); only custom transports are affected. + See :meth:`Transport.subject_id_modulus`. """ from ._node import NodeImpl diff --git a/src/pycyphal2/_node.py b/src/pycyphal2/_node.py index f3f19be40..ed73f7ba4 100644 --- a/src/pycyphal2/_node.py +++ b/src/pycyphal2/_node.py @@ -28,7 +28,7 @@ ScoutHeader, deserialize_header, ) -from ._transport import SubjectWriter, Transport, TransportArrival +from ._transport import SUBJECT_ID_MODULUS_16bit, SubjectWriter, Transport, TransportArrival from ._api import Topic, Node, Publisher, Subscriber, Breadcrumb, Closable, ClosedError, Instant, Priority, SendError from ._api import SUBJECT_ID_PINNED_MAX @@ -48,6 +48,7 @@ ACK_TX_TIMEOUT = 1.0 SESSION_LIFETIME = 60.0 IMPLICIT_TOPIC_TIMEOUT = 600.0 +HOUSEKEEPING_PERIOD = 1.0 # Stale-state sweep cadence; well inside SESSION_LIFETIME. REORDERING_CAPACITY = 16 ASSOC_SLACK_LIMIT = 2 DEDUP_HISTORY = 512 @@ -95,7 +96,7 @@ def _name_consume_pin_suffix(name: str) -> tuple[str, int | None]: if ch == "#": hash_pos = i break - if not ch.isdigit(): + if not "0" <= ch <= "9": # ASCII only: str.isdigit() accepts chars that int() rejects, e.g. '²' return (name, None) if hash_pos < 0: return (name, None) @@ -122,20 +123,23 @@ def _name_is_homeful(name: str) -> bool: return name == "~" or name.startswith("~/") +def _name_has_pattern_tokens(name: str) -> bool: + """A name is a pattern iff a whole '/'-segment is a substitution token ('*' or '>'). Tokens inside a longer + segment (e.g. 'ab*cd') are literal, as in the reference classifier (wkv_has_substitution_tokens).""" + return any(seg in ("*", ">") for seg in name.split("/")) + + def _is_valid_wire_name(name: str) -> bool: - """True if `name` is a well-formed *resolved* wire topic name, as required of names received in gossip: - nonempty, length-bounded, printable ASCII (33-126), already normalized (no leading/trailing/duplicate - '/'), verbatim (no '*'/'>' pattern tokens), not homeful ('~'/'~/...'), and pin-free (no '#' suffix). - The last two are stripped/expanded by resolve_name before a name reaches the wire, so their presence - means the gossip is unresolved/non-canonical and must not create a local topic.""" + """True if `name` is a well-formed *resolved* wire topic name, as required of names received in gossip. + Homeful ('~'/'~/...') and pinned ('#') forms are stripped/expanded by resolve_name before a name + reaches the wire, so their presence means the gossip is unresolved and must not create a local topic.""" return ( bool(name) and len(name) <= TOPIC_NAME_MAX - and "*" not in name - and ">" not in name + and not _name_has_pattern_tokens(name) and not _name_is_homeful(name) - and _name_consume_pin_suffix(name)[1] is None and all(33 <= ord(ch) <= 126 for ch in name) + and _name_consume_pin_suffix(name)[1] is None and _name_normalize(name) == name ) @@ -184,7 +188,7 @@ def resolve_name( if o < 33 or o > 126: raise ValueError(f"Invalid character in name: {ch!r}") - verbatim = "*" not in resolved and ">" not in resolved + verbatim = not _name_has_pattern_tokens(resolved) if pin is not None and not verbatim: raise ValueError("Pattern names cannot be pinned") return resolved, pin, verbatim @@ -219,12 +223,29 @@ def match_pattern(pattern: str, name: str) -> list[tuple[str, int]] | None: return subs +def is_valid_subject_id_modulus(modulus: int) -> bool: + """The reference predicate (cy.c is_valid_subject_id_modulus): at least 57203, prime, and ≡ 3 (mod 4). + The quadratic probe (hash + evictions²) mod m covers the residue space only under these conditions; + a degenerate modulus would make the synchronous displacement loop in topic_allocate effectively + non-terminating, hard-blocking the event loop.""" + # The reference modulus is a uint32; that bound also caps the trial division at ~65536 iterations, so an + # untrusted custom-transport value cannot hang the primality test. + if modulus < SUBJECT_ID_MODULUS_16bit or modulus > 0xFFFFFFFF or modulus % 4 != 3: + return False + d = 3 + while d * d <= modulus: + if modulus % d == 0: + return False + d += 2 + return True + + def compute_subject_id(topic_hash: int, evictions: int, modulus: int) -> int: if evictions >= EVICTIONS_PINNED_MIN: return 0xFFFFFFFF - evictions - h = topic_hash % modulus - e = evictions % modulus - return SUBJECT_ID_PINNED_MAX + 1 + ((h + ((e * e) % modulus)) % modulus) + # The sum wraps mod 2**64 before the reduction, matching the reference uint64 arithmetic; without the + # wrap, a large hash plus a near-boundary eviction count would diverge from C on the subject-ID. + return SUBJECT_ID_PINNED_MAX + 1 + (((topic_hash + evictions * evictions) & U64_MASK) % modulus) @dataclass @@ -232,9 +253,13 @@ class Association: """Tracks a known remote subscriber for reliable delivery ACK tracking.""" remote_id: int + # Diagnostics only, deliberately NOT an eviction input, matching the reference association_t ("Not used + # for eviction, only for diagnostics and possibly API exposure", cy.c). Eviction is driven by `slack`. last_seen: float slack: int = 0 seqno_witness: int = 0 + # An association cannot be dropped while a publish tracker still references it (reference parity: + # "The association cannot be removed unless zero to avoid dangly pointers", cy.c). pending_count: int = 0 @@ -383,6 +408,10 @@ def __init__(self, node: NodeImpl, name: str, evictions: int, now: float) -> Non self.sub_listener: Closable | None = None self.couplings: list[Coupling] = [] self.is_implicit = True + # Unbounded by design, matching the reference, which carries an explicit TODO for the same gap + # ("there should be a limit on the number of associations to prevent DoS ... ~500 might be a + # reasonable default", cy.c). Bounding it ahead of the reference would diverge on reliable-delivery + # accounting, so the fix belongs upstream first. self.associations: dict[int, Association] = {} self.dedup: dict[int, DedupState] = {} self.publish_futures: dict[int, PublishTracker] = {} @@ -447,7 +476,13 @@ def ensure_writer(self) -> SubjectWriter: def ensure_listener(self) -> None: if self.sub_listener is None and self.couplings: sid = self.subject_id(self._node.transport.subject_id_modulus) - self.sub_listener = self._node.acquire_subject_listener(self, sid) + # Repair model (cy.c topic_sync_subject_reader): an acquisition failure is left for the next + # opportunity (topic sync, periodic gossip) to retry, which keeps sync_listener() infallible. + try: + self.sub_listener = self._node.acquire_subject_listener(self, sid) + except Exception as ex: + _logger.error("Listener acquisition failed for '%s' sid=%d; will retry: %s", self._name, sid, ex) + return _logger.info("Listener acquired for '%s' sid=%d", self._name, sid) def sync_listener(self) -> None: @@ -469,12 +504,24 @@ def release_transport_handles(self) -> None: def compute_is_implicit(self) -> bool: has_verbatim_sub = any(not c.root.is_pattern for c in self.couplings) - return self.pub_count == 0 and not has_verbatim_sub + # An open response stream or in-flight reliable publish keeps the topic explicit so implicit GC + # cannot destroy it (closed "zombie" streams awaiting their dedup-cleanup timer do not count). + # Needed because, unlike the C API, Python lets a stream/publish outlive its publisher. + has_open_stream = any(not s.closed for s in self.request_futures.values()) + has_pending_publish = bool(self.publish_futures) + return self.pub_count == 0 and not has_verbatim_sub and not has_open_stream and not has_pending_publish def sync_implicit(self) -> None: """Sync implicitness and transport state with the reference state machine.""" self._node.sync_topic_lifecycle(self) + def drop_stale_dedup(self, now: float) -> None: + """Aggregate sweep of per-remote dedup state (reference dedup_drop_stale). The per-arrival prune only + touches the arriving remote; this retires departed ones so the map cannot grow without bound.""" + stale = [rid for rid, st in self.dedup.items() if (st.last_active + SESSION_LIFETIME) < now] + for rid in stale: + del self.dedup[rid] + def log_age(origin: float, now: float) -> int: diff = int(now - origin) @@ -514,6 +561,11 @@ def __init__(self, transport: Transport, *, home: str, namespace: str) -> None: self.respond_futures: dict[tuple[int, ...], RespondTracker] = {} modulus = transport.subject_id_modulus + if not is_valid_subject_id_modulus(modulus): + raise ValueError( + f"Invalid subject_id_modulus {modulus}: " + f"must be at least {SUBJECT_ID_MODULUS_16bit}, prime, and congruent to 3 modulo 4" + ) sid_max = SUBJECT_ID_PINNED_MAX + modulus self.broadcast_subject_id = (1 << (int(math.log2(sid_max)) + 1)) - 1 self.gossip_shard_count = self.broadcast_subject_id - (sid_max + 1) @@ -524,18 +576,30 @@ def __init__(self, transport: Transport, *, home: str, namespace: str) -> None: def broadcast_handler(arrival: TransportArrival) -> None: self.on_subject_arrival(self.broadcast_subject_id, arrival) - self.broadcast_listener = transport.subject_listen(self.broadcast_subject_id, broadcast_handler) + try: + self.broadcast_listener = transport.subject_listen(self.broadcast_subject_id, broadcast_handler) + except BaseException: + self.broadcast_writer.close() # Roll back the writer so a retry does not see a duplicate. + raise self.gossip_shard_writers: dict[int, SubjectWriter] = {} self.gossip_shard_listeners: dict[int, Closable] = {} self.shared_subject_writers: dict[int, SharedSubjectWriter] = {} self.shared_subject_listeners: dict[int, SharedSubjectListener] = {} - transport.unicast_listen(self.on_unicast_arrival) + # Nobody closes the transport on constructor failure, so an unguarded raise here would leak the + # broadcast writer and listener. Only third-party transports can raise here; the built-in ones cannot. + try: + transport.unicast_listen(self.on_unicast_arrival) + except BaseException: + self.broadcast_listener.close() + self.broadcast_writer.close() + raise self._implicit_topics: OrderedDict[TopicImpl, None] = OrderedDict() self._implicit_gc_wakeup = asyncio.Event() self._gc_task = self.loop.create_task(self.implicit_gc_loop()) + self._housekeeping_task = self.loop.create_task(self._housekeeping_loop()) _logger.info( "Node init home='%s' ns='%s' broadcast_sid=%d shards=%d", @@ -597,9 +661,11 @@ def advertise(self, name: str) -> Publisher: if not verbatim: raise ValueError("Cannot advertise on a pattern name") topic = self.topic_ensure(resolved, pin) + # Acquire the fallible writer before mutating publish state, so a failure leaves the topic as an + # ordinary implicit topic (GC'd) rather than a phantom publisher with no writer. + topic.ensure_writer() topic.pub_count += 1 topic.sync_implicit() - topic.ensure_writer() _logger.info( "Advertise '%s' -> '%s' sid=%d", name, @@ -616,24 +682,25 @@ def subscribe(self, name: str, *, reordering_window: float | None = None) -> Sub if pin is not None and not verbatim: raise ValueError("Pattern names cannot be pinned") - if verbatim: - root = self.sub_roots_verbatim.get(resolved) - if root is None: - root = SubscriberRoot(name=resolved, is_pattern=False) - self.sub_roots_verbatim[resolved] = root - else: - root = self.sub_roots_pattern.get(resolved) - if root is None: - root = SubscriberRoot(name=resolved, is_pattern=True, needs_scouting=True) - self.sub_roots_pattern[resolved] = root - + # Acquire the fallible resources first — the subscriber (reordering-window validation) and, for a + # verbatim name, its topic — so a failure leaves no half-registered root or subscriber behind. + registry = self.sub_roots_verbatim if verbatim else self.sub_roots_pattern + root = registry.get(resolved) + new_root = root is None + if root is None: + root = SubscriberRoot(name=resolved, is_pattern=not verbatim, needs_scouting=not verbatim) subscriber = SubscriberImpl(self, root, resolved, verbatim, reordering_window) + verbatim_topic = self.topic_ensure(resolved, pin) if verbatim else None + + # Commit — everything below is infallible. + if new_root: + registry[resolved] = root root.subscribers.append(subscriber) if verbatim: - topic = self.topic_ensure(resolved, pin) - self.couple_topic_root(topic, root) - topic.sync_implicit() + assert verbatim_topic is not None + self.couple_topic_root(verbatim_topic, root) + verbatim_topic.sync_implicit() else: for topic in list(self.topics_by_name.values()): self.couple_topic_root(topic, root) @@ -683,12 +750,17 @@ def topic_ensure(self, name: str, pin: int | None) -> TopicImpl: topic = TopicImpl(self, name, evictions, now) self.topics_by_name[name] = topic self.topics_by_hash[topic.hash] = topic - self.ensure_gossip_shard(self.gossip_shard_subject_id(topic.hash)) - self.touch_implicit_topic(topic) - self.topic_allocate(topic, evictions, now) - for root in self.sub_roots_pattern.values(): - self.couple_topic_root(topic, root) - topic.sync_listener() + # Index first so that the rollback primitive (destroy_topic) can find and undo the topic. + try: + self.ensure_gossip_shard(self.gossip_shard_subject_id(topic.hash)) + self.touch_implicit_topic(topic) + self.topic_allocate(topic, evictions, now) + for root in self.sub_roots_pattern.values(): + self.couple_topic_root(topic, root) + topic.sync_listener() + except BaseException: + self._rollback_topic(name) + raise self.notify_implicit_gc() _logger.info( "Topic created '%s' hash=%016x sid=%d", @@ -698,6 +770,13 @@ def topic_ensure(self, name: str, pin: int | None) -> TopicImpl: ) return topic + def _rollback_topic(self, name: str) -> None: + """Undo a partially-built topic without masking the exception that triggered the rollback.""" + try: + self.destroy_topic(name) + except Exception: + _logger.exception("Rollback of partially-built topic '%s' failed", name) + def topic_allocate(self, topic: TopicImpl, new_evictions: int, now: float) -> None: """Iterative subject-ID allocation with collision resolution. Mirrors topic_allocate() in cy.c.""" modulus = self.transport.subject_id_modulus @@ -734,6 +813,10 @@ def topic_allocate(self, topic: TopicImpl, new_evictions: int, now: float) -> No del self.topics_by_subject_id[new_sid] self.topics_by_subject_id[new_sid] = t if collider.pub_writer is not None: + # Winner acquires BEFORE the loser releases below, so the shared handle survives the + # handover on its refcount ("the winner acquires first, then the loser releases" in the + # reference). It is therefore a pure refcount bump on the existing registry entry: no + # transport call, so the cascade cannot fail midway and strand a displaced topic. t.pub_writer = self.acquire_subject_writer(t, new_sid) t.sync_listener() self.schedule_gossip_urgent(t) @@ -743,6 +826,11 @@ def topic_allocate(self, topic: TopicImpl, new_evictions: int, now: float) -> No work.append((t, ev + 1)) def sync_topic_lifecycle(self, topic: TopicImpl) -> None: + # Reachable after close(): a disposed response stream's cancelled publish task re-syncs implicitness + # from its finally clause on a later loop iteration. Without this guard that tail would spawn an + # uncancellable gossip task on a dead node and call subject_listen on a closed transport. + if self._closed: + return implicit = topic.compute_is_implicit() if implicit != topic.is_implicit: topic.is_implicit = implicit @@ -837,7 +925,12 @@ def ensure_gossip_shard(self, shard_sid: int) -> SubjectWriter: def handler(arrival: TransportArrival) -> None: self.on_subject_arrival(shard_sid, arrival) - self.gossip_shard_listeners[shard_sid] = self.transport.subject_listen(shard_sid, handler) + try: + self.gossip_shard_listeners[shard_sid] = self.transport.subject_listen(shard_sid, handler) + except BaseException: + self.gossip_shard_writers.pop(shard_sid, None) # Roll back the just-created writer. + writer.close() + raise _logger.debug("Gossip shard writer/listener for sid=%d", shard_sid) return writer @@ -947,6 +1040,8 @@ async def _gossip_event_urgent(self, topic: TopicImpl) -> None: await self.send_gossip(topic, broadcast=True) async def _gossip_event_periodic(self, topic: TopicImpl) -> None: + # Retry a previously-failed listener acquisition on the gossip cadence (reference topic_sync_subject_reader). + topic.sync_listener() self._reschedule_gossip_periodic(topic, suppressed=False) broadcast = (topic.gossip_counter < GOSSIP_BROADCAST_RATIO) or ( (topic.gossip_counter % GOSSIP_BROADCAST_RATIO) == 0 @@ -1372,12 +1467,19 @@ def topic_subscribe_if_matching( topic.ts_origin = now - lage_to_seconds(lage) self.topics_by_name[name] = topic self.topics_by_hash[topic_hash] = topic - self.ensure_gossip_shard(self.gossip_shard_subject_id(topic.hash)) - self.touch_implicit_topic(topic) - self.topic_allocate(topic, evictions, now) - for root in matches: - self.couple_topic_root(topic, root) - topic.sync_listener() + # Wire-driven path: must never raise (untrusted input). A mid-setup failure rolls the topic back + # and drops the gossip; a later gossip retries. + try: + self.ensure_gossip_shard(self.gossip_shard_subject_id(topic.hash)) + self.touch_implicit_topic(topic) + self.topic_allocate(topic, evictions, now) + for root in matches: + self.couple_topic_root(topic, root) + topic.sync_listener() + except Exception as ex: + self._rollback_topic(name) + _logger.warning("Implicit topic '%s' setup failed, dropped: %s", name, ex) + return None self.notify_implicit_gc() _logger.info("Implicit topic '%s' created from gossip", name) return topic @@ -1400,12 +1502,11 @@ def notify_implicit_gc(self) -> None: if not self._closed: self._implicit_gc_wakeup.set() - def _next_implicit_gc_delay(self, now: float | None = None) -> float | None: - now = time.monotonic() if now is None else now + def _next_implicit_gc_delay(self) -> float | None: if not self._implicit_topics: return None oldest = next(reversed(self._implicit_topics)) - return max(0.0, (oldest.ts_animated + IMPLICIT_TOPIC_TIMEOUT) - now) + return max(0.0, (oldest.ts_animated + IMPLICIT_TOPIC_TIMEOUT) - time.monotonic()) def _retire_one_expired_implicit_topic(self, now: float) -> bool: if not self._implicit_topics: @@ -1417,6 +1518,20 @@ def _retire_one_expired_implicit_topic(self, now: float) -> bool: _logger.info("GC removed implicit topic '%s'", oldest.name) return True + def sweep_stale_states(self, now: float) -> None: + """Time-driven retirement of per-remote dedup and reordering state, so neither grows without bound + after the traffic that created it stops (reference poll's round-robin dedup_drop_stale / + reordering_drop_stale, done sweep-all here).""" + from ._subscriber import SubscriberImpl + + for topic in list(self.topics_by_name.values()): + topic.drop_stale_dedup(now) + for registry in (self.sub_roots_verbatim, self.sub_roots_pattern): + for root in list(registry.values()): + for sub in list(root.subscribers): + if isinstance(sub, SubscriberImpl): + sub.drop_stale_reordering(now) + async def implicit_gc_loop(self) -> None: try: while not self._closed: @@ -1431,7 +1546,27 @@ async def implicit_gc_loop(self) -> None: continue except asyncio.TimeoutError: pass - self._retire_one_expired_implicit_topic(time.monotonic()) + try: + self._retire_one_expired_implicit_topic(time.monotonic()) + except Exception: + # A faulty retirement must not permanently disable implicit-topic GC. Back off first: + # an expired topic yields a zero delay, so an immediate retry would spin on a hard fault. + _logger.exception("Implicit topic retirement failed; backing off") + await asyncio.sleep(HOUSEKEEPING_PERIOD) + except asyncio.CancelledError: + pass + + async def _housekeeping_loop(self) -> None: + # Dedicated task so the sweep keeps a steady cadence regardless of implicit-GC activity. + try: + while not self._closed: + await asyncio.sleep(HOUSEKEEPING_PERIOD) + try: + self.sweep_stale_states(time.monotonic()) + except Exception: + # This loop is the only bound on per-remote state growth, so it must outlive a faulty + # sweep; an escaping exception would kill the task silently and leave the node unbounded. + _logger.exception("Stale-state sweep failed; continuing") except asyncio.CancelledError: pass @@ -1439,6 +1574,10 @@ def destroy_topic(self, name: str) -> None: topic = self.topics_by_name.get(name) if topic is None: return + # Dispose response streams before discard_implicit_topic, else a re-touch of the implicit list + # could resurrect the topic mid-destroy. + for stream in list(topic.request_futures.values()): + stream.dispose() if topic.gossip_task is not None: self._cancel_gossip(topic) self.discard_implicit_topic(topic) @@ -1453,6 +1592,7 @@ def destroy_topic(self, name: str) -> None: topic.associations.clear() topic.dedup.clear() topic.publish_futures.clear() + topic.request_futures.clear() self.notify_implicit_gc() _logger.info("Topic destroyed '%s'", name) @@ -1462,12 +1602,17 @@ def close(self) -> None: self._closed = True _logger.info("Node closing home='%s'", self._home) # Unblock anything awaiting on a subscriber (`async for`): closing each enqueues StopAsyncIteration, - # otherwise a default (no-liveness-timeout) subscriber would wait on its queue forever. (Reliable - # publishes / response streams are deadline-bounded and resolve on their own.) + # otherwise a default (no-liveness-timeout) subscriber would wait on its queue forever. for root in list(self.sub_roots_verbatim.values()) + list(self.sub_roots_pattern.values()): for sub in list(root.subscribers): sub.close() + # Likewise for response streams: a far-off (or infinite) response timeout would otherwise keep the + # retry task and payload graph alive against the closed transport until that timeout. + for topic in list(self.topics_by_name.values()): + for stream in list(topic.request_futures.values()): + stream.dispose() self._gc_task.cancel() + self._housekeeping_task.cancel() for root in list(self.sub_roots_pattern.values()): if root.scout_task is not None: root.scout_task.cancel() @@ -1486,8 +1631,10 @@ def close(self) -> None: self.shared_subject_listeners.clear() for w in self.gossip_shard_writers.values(): w.close() + self.gossip_shard_writers.clear() for gossip_listener in self.gossip_shard_listeners.values(): gossip_listener.close() + self.gossip_shard_listeners.clear() self._monitor_callbacks.clear() self._implicit_topics.clear() self.transport.close() diff --git a/src/pycyphal2/_publisher.py b/src/pycyphal2/_publisher.py index 231843820..34d10a1d4 100644 --- a/src/pycyphal2/_publisher.py +++ b/src/pycyphal2/_publisher.py @@ -112,6 +112,10 @@ async def request( if self.closed: raise SendError("Publisher closed") + response_timeout = float(response_timeout) + if math.isnan(response_timeout) or response_timeout < 0.0: + raise ValueError("response_timeout must be non-negative (or +inf to disable liveness)") + tag = self._topic.next_tag() payload = bytes(message) @@ -124,17 +128,18 @@ async def request( ) self._topic.request_futures[tag] = stream - tracker = self._prepare_reliable_publish_tracker(tag) + # Any failure past registration must close() the stream, not just pop the tag; otherwise a + # concurrently closing publisher could leave the topic permanently explicit. + tracker: PublishTracker | None = None try: + tracker = self._prepare_reliable_publish_tracker(tag) initial_window = await self._reliable_publish_start(delivery_deadline, tag, payload, tracker) - except asyncio.CancelledError: - tracker.compromised = True - self._topic.request_futures.pop(tag, None) - self._release_reliable_publish_tracker(tag, tracker) - raise - except BaseException: - self._topic.request_futures.pop(tag, None) - self._release_reliable_publish_tracker(tag, tracker) + except BaseException as ex: + if tracker is not None: + if isinstance(ex, asyncio.CancelledError): + tracker.compromised = True # publish_tracker_release reads this, so set it before release. + self._release_reliable_publish_tracker(tag, tracker) + stream.close() raise task = self._node.loop.create_task( @@ -192,6 +197,9 @@ def _prepare_reliable_publish_tracker(self, tag: int) -> PublishTracker: def _release_reliable_publish_tracker(self, tag: int, tracker: PublishTracker) -> None: self._topic.publish_futures.pop(tag, None) self._node.publish_tracker_release(self._topic, tracker) + # A topic held explicit only by this publish can now become implicit and be GC'd; without this + # it would gossip forever. + self._topic.sync_implicit() async def _send_reliable_publish( self, @@ -311,6 +319,9 @@ def __init__( self._response_timeout = response_timeout self.queue: asyncio.Queue[Response | BaseException] = asyncio.Queue() self.closed = False + # Never pruned during the stream's life, matching the reference request_future_t.remote_by_id + # ("States are never removed assuming that futures are short-lived and/or the responder set is + # mostly constant", cy.c). close() retains it to re-ack late duplicates via the zombie timer. self._reliable_remote_by_id: dict[int, ResponseRemoteState] = {} self._publish_task: asyncio.Task[None] | None = None self._cleanup_handle: asyncio.TimerHandle | None = None @@ -321,8 +332,10 @@ def __aiter__(self) -> ResponseStreamImpl: async def __anext__(self) -> Response: if self.closed: raise StopAsyncIteration + # inf disables liveness (waits forever), mirroring Subscriber.timeout. + timeout = self._response_timeout if self._response_timeout != float("inf") else None try: - item = await asyncio.wait_for(self.queue.get(), timeout=self._response_timeout) + item = await asyncio.wait_for(self.queue.get(), timeout=timeout) except asyncio.TimeoutError: raise LivenessError("Response timeout") if isinstance(item, StopAsyncIteration): @@ -404,4 +417,20 @@ def close(self) -> None: else: self._remove_from_topic() self.queue.put_nowait(StopAsyncIteration()) + # The stream no longer keeps the topic explicit, so it can now be GC'd instead of gossiping + # forever. Safe during node teardown -- notify_implicit_gc() is a no-op when the node is closed. + self._topic.sync_implicit() _logger.debug("Response stream closed for tag=%d", self._message_tag) + + def dispose(self) -> None: + """Force-remove the stream during node/topic teardown. Unlike close(), retains no zombie and does + not re-sync implicitness, the topic being torn down anyway.""" + was_open = not self.closed + self.closed = True + if self._publish_task is not None: + self._publish_task.cancel() + self._publish_task = None + self._remove_from_topic() + self._reliable_remote_by_id.clear() + if was_open: + self.queue.put_nowait(StopAsyncIteration()) diff --git a/src/pycyphal2/_subscriber.py b/src/pycyphal2/_subscriber.py index 519a77c49..ef612514b 100644 --- a/src/pycyphal2/_subscriber.py +++ b/src/pycyphal2/_subscriber.py @@ -114,7 +114,7 @@ def deliver(self, arrival: Arrival, tag: int, remote_id: int) -> bool: if self._reordering_window is None: self.queue.put_nowait(arrival) return True - self._drop_stale_reordering(arrival.timestamp.s) + self.drop_stale_reordering(arrival.timestamp.s) topic_hash = arrival.breadcrumb.topic.hash key = (remote_id, topic_hash) state = self._reordering.get(key) @@ -210,7 +210,7 @@ def on_timeout() -> None: state.timeout_handle = loop.call_later(delay, on_timeout) - def _drop_stale_reordering(self, now: float) -> None: + def drop_stale_reordering(self, now: float) -> None: stale = [key for key, state in self._reordering.items() if (state.last_active_at + SESSION_LIFETIME) < now] for key in stale: state = self._reordering.pop(key) diff --git a/src/pycyphal2/_transport.py b/src/pycyphal2/_transport.py index 02f2995d9..cf8aa45fa 100644 --- a/src/pycyphal2/_transport.py +++ b/src/pycyphal2/_transport.py @@ -50,7 +50,9 @@ class Transport(Closable): @abstractmethod def subject_id_modulus(self) -> int: """ - Constant, cannot be changed while the transport is in used because that would invalidate subject allocations. + Constant, cannot be changed while the transport is in use because that would invalidate subject allocations. + Must satisfy the reference predicate — at least ``SUBJECT_ID_MODULUS_16bit`` (57203), prime, congruent to 3 + modulo 4 — otherwise :meth:`pycyphal2.Node.new` rejects the transport with ``ValueError``. """ raise NotImplementedError diff --git a/src/pycyphal2/can/_interface.py b/src/pycyphal2/can/_interface.py index a4f461dee..d0e737d75 100644 --- a/src/pycyphal2/can/_interface.py +++ b/src/pycyphal2/can/_interface.py @@ -9,7 +9,6 @@ from .. import Closable, ClosedError, Instant CAN_EXT_ID_MASK = (1 << 29) - 1 -CAN_STD_ID_MASK = (1 << 11) - 1 def closed_error(interface: str, failure: BaseException | None) -> ClosedError: diff --git a/src/pycyphal2/can/_media_slcan.py b/src/pycyphal2/can/_media_slcan.py index 9420fac49..6daaa52c6 100644 --- a/src/pycyphal2/can/_media_slcan.py +++ b/src/pycyphal2/can/_media_slcan.py @@ -6,7 +6,7 @@ import logging -from ._interface import CAN_EXT_ID_MASK, CAN_STD_ID_MASK, Frame +from ._interface import CAN_EXT_ID_MASK, Frame from ._wire import DLC_TO_LENGTH, MTU_CAN_CLASSIC _logger = logging.getLogger(__name__) @@ -126,12 +126,12 @@ def _parse_line(line: bytes) -> Frame | None: command = line[:1] if command in (b"T", b"x"): return _parse_data_frame(line, id_length=8, max_payload_length=MTU_CAN_CLASSIC) - if command == b"t": - return _parse_data_frame(line, id_length=3, max_payload_length=MTU_CAN_CLASSIC) if command == b"D": return _parse_data_frame(line, id_length=8, max_payload_length=64) - if command in (b"r", b"R"): - _logger.debug("SLCAN drop unsupported frame type cmd=%r", command) + if command in (b"t", b"r", b"R"): + # The Interface contract is extended-data-only and Frame carries no IDE/RTR discriminator, so a + # standard-ID 't'/'r' would alias an extended frame with a small ID; RTR has no Cyphal meaning. + _logger.debug("SLCAN drop unusable frame type cmd=%r", command) return None _logger.debug("SLCAN drop unknown line=%r", line) return None @@ -155,9 +155,6 @@ def _parse_data_frame(line: bytes, *, id_length: int, max_payload_length: int) - if len(line) < expected: _logger.debug("SLCAN drop data dlc mismatch len=%d expected=%d", len(line), expected) return None - if id_length == 3 and identifier > CAN_STD_ID_MASK: - _logger.debug("SLCAN drop invalid standard id=%x", identifier) - return None data = _parse_hex_bytes(line[header_length:expected]) if data is None: _logger.debug("SLCAN drop malformed data id=%08x", identifier) diff --git a/src/pycyphal2/can/_reassembly.py b/src/pycyphal2/can/_reassembly.py index 3e6b5bb3a..adca64f06 100644 --- a/src/pycyphal2/can/_reassembly.py +++ b/src/pycyphal2/can/_reassembly.py @@ -63,13 +63,17 @@ class Endpoint: class Reassembler: @staticmethod def cleanup_sessions(endpoints: Iterable[Endpoint], now_ns: int) -> None: + # A slot-free session dies at the transfer-ID timeout, not the slot retention timeout, as in the + # reference (canard_poll): the admission logic already treats it as stale, and retaining it longer + # only delays cross-interface transfer-ID reuse after a redundant failover. stale_deadline = now_ns - RX_SESSION_RETENTION_NS + idle_deadline = now_ns - TRANSFER_ID_TIMEOUT_NS for endpoint in endpoints: for source_id, session in list(endpoint.sessions.items()): for priority, slot in enumerate(session.slots): if slot is not None and slot.start_ts_ns < stale_deadline: session.slots[priority] = None - if all(slot is None for slot in session.slots) and session.last_admission_ts_ns < stale_deadline: + if all(slot is None for slot in session.slots) and session.last_admission_ts_ns < idle_deadline: endpoint.sessions.pop(source_id, None) @staticmethod diff --git a/src/pycyphal2/can/_transport.py b/src/pycyphal2/can/_transport.py index 30d19c74f..f02099863 100644 --- a/src/pycyphal2/can/_transport.py +++ b/src/pycyphal2/can/_transport.py @@ -250,7 +250,9 @@ def unicast_listen(self, handler: Callable[[TransportArrival], None]) -> None: async def unicast(self, deadline: Instant, priority: Priority, remote_id: int, message: bytes | memoryview) -> None: if self._closed: raise ClosedError("CAN transport closed") - if not (1 <= remote_id <= NODE_ID_MAX): + # Node-ID 0 is a regular node in Cyphal/CAN v1 (only v0 treated it as anonymous); rejecting it + # would leave the ACK path unable to answer a node-0 peer. + if not (0 <= remote_id <= NODE_ID_MAX): raise ValueError(f"Invalid remote node-ID: {remote_id}") transfer_id = self._unicast_tid[remote_id] self._unicast_tid[remote_id] = (transfer_id + 1) % TRANSFER_ID_MODULO @@ -365,7 +367,11 @@ async def _reader_loop(self, itf: Interface) -> None: iface_index = self._interface_index.get(id(itf)) if iface_index is None: return - self._ingest_frame(iface_index, frame) + try: + self._ingest_frame(iface_index, frame) + except Exception: + # A raising handler must not kill the reader loop; drop the frame and keep serving. + _logger.exception("Frame ingest raised iface=%s", itf.name) def _drop_interface(self, itf: Interface, ex: BaseException) -> None: if itf not in self._interfaces: @@ -465,7 +471,11 @@ async def _cleanup_loop(self) -> None: try: while not self._closed: await asyncio.sleep(1.0) - Reassembler.cleanup_sessions(self._endpoints.values(), Instant.now().ns) + try: + Reassembler.cleanup_sessions(self._endpoints.values(), Instant.now().ns) + except Exception: + # Retirement is not traffic-driven: a faulty sweep must not kill the loop and leak sessions. + _logger.exception("Session cleanup failed; continuing") except asyncio.CancelledError: raise diff --git a/src/pycyphal2/can/pythoncan.py b/src/pycyphal2/can/pythoncan.py index 9597b5d1b..5a59bda92 100644 --- a/src/pycyphal2/can/pythoncan.py +++ b/src/pycyphal2/can/pythoncan.py @@ -112,12 +112,12 @@ def purge(self) -> None: async def receive(self) -> TimestampedFrame: self._raise_if_closed() - while True: - item = await self._rx_queue.get() - if isinstance(item, BaseException): - self._fail(item) - raise ClosedError(f"PythonCAN interface {self._name} receive failed") from item - return item + item = await self._rx_queue.get() + if isinstance(item, BaseException): + # The failure was already recorded by _fail() and folded into this sentinel; feeding it back + # through _fail() here would misrecord a clean close as an interface failure. As in SocketCAN. + raise item + return item def close(self) -> None: with self._admin_lock: @@ -129,9 +129,12 @@ def close(self) -> None: self._tx_task.cancel() self._tx_task = None try: - self._rx_queue.put_nowait(ClosedError(f"PythonCAN interface {self._name} closed")) + # Carries self._failure as the cause when close() came via _fail(), so a parked reader + # learns why the interface died rather than just that it closed. + self._rx_queue.put_nowait(self._closed_error()) except Exception: - pass + # Never silent: without the sentinel a parked reader hangs forever. + _logger.exception("PythonCAN could not install the terminal RX sentinel on %s", self._name) try: self._bus.shutdown() except Exception as ex: @@ -159,12 +162,14 @@ async def _tx_loop(self) -> None: _logger.debug("PythonCAN tx drop expired iface=%s id=%08x", self._name, entry.id) job.abort(SendError(f"PythonCAN interface {self._name} tx deadline expired")) continue + # The FD flag follows the interface mode, not the payload length; see the REFERENCE PARITY + # note in socketcan._encode on why BRS is always set on an FD interface. msg = can.Message( arbitration_id=entry.id, is_extended_id=True, data=entry.payload, - is_fd=self._fd and len(entry.payload) > 8, - bitrate_switch=self._fd and len(entry.payload) > 8, + is_fd=self._fd, + bitrate_switch=self._fd, ) try: await asyncio.wait_for(loop.run_in_executor(None, self._bus.send, msg, timeout), timeout=timeout) @@ -195,7 +200,9 @@ def _rx_thread_func(self) -> None: except Exception as ex: if not self._closed: try: - self._loop.call_soon_threadsafe(self._rx_queue.put_nowait, ex) + # _fail() stores the cause and closes, installing the terminal sentinel; this + # also propagates the failure when nobody is parked in receive(). + self._loop.call_soon_threadsafe(self._fail, ex) except RuntimeError: pass return diff --git a/src/pycyphal2/can/socketcan.py b/src/pycyphal2/can/socketcan.py index 4c2cb42b5..cdd395e80 100644 --- a/src/pycyphal2/can/socketcan.py +++ b/src/pycyphal2/can/socketcan.py @@ -22,7 +22,10 @@ _CAN_FILTER_CAPACITY = 64 _CAN_INTERFACE_TYPE = 280 -_CANFD_FDF = getattr(socket, "CANFD_FDF", 0) +# CAN FD flag bits from linux/can.h, hardcoded because CPython's socket module does not expose them on +# any supported version and a getattr() fallback would silently clear them. +_CANFD_BRS = 0x01 # Bit-rate switch: the data phase runs at the higher FD bit rate. +_CANFD_FDF = 0x04 # Marks the frame as CAN FD in the dual-use struct canfd_frame. _CAN_FRAME_STRUCT = struct.Struct("=IB3x8s") _CANFD_FRAME_STRUCT = struct.Struct("=IBBBB64s") _CAN_FILTER_STRUCT = struct.Struct("=II") @@ -47,6 +50,11 @@ def __init__(self, name: str) -> None: self._failure: BaseException | None = None self._tx = TxQueue() self._tx_task: asyncio.Task[None] | None = None + # RX runs in its own task feeding a queue so close()/fail() can wake a parked reader with a + # sentinel; the selector loop never wakes a pending sock_recv on a bare socket.close(). + # Mirrors the python-can and webserial backends. + self._rx_queue: asyncio.Queue[TimestampedFrame | BaseException] = asyncio.Queue() + self._rx_task: asyncio.Task[None] | None = None @property def name(self) -> str: @@ -88,28 +96,63 @@ def purge(self) -> None: async def receive(self) -> TimestampedFrame: self._raise_if_closed() + if self._rx_task is None: + self._rx_task = asyncio.get_running_loop().create_task(self._rx_loop()) + self._rx_task.add_done_callback(self._on_task_done) + item = await self._rx_queue.get() + if isinstance(item, BaseException): + # The sentinel is either the failure recorded by _fail() or a plain ClosedError from an + # explicit close; raise it as is so a clean close is not misrecorded as a failure. + raise item + return item + + async def _rx_loop(self) -> None: loop = asyncio.get_running_loop() recv_size = _FD_FRAME_SIZE if self._fd else _CLASSIC_FRAME_SIZE - while True: + while not self._closed: try: raw = await loop.sock_recv(self._sock, recv_size) except asyncio.CancelledError: raise except OSError as ex: - self._fail(ex) - raise ClosedError(f"SocketCAN interface {self._name} receive failed") from ex + if not self._closed: + self._fail(ex) # Records the failure, closes, and installs the terminal sentinel. + return frame = self._decode(raw) if frame is not None: - return frame + self._rx_queue.put_nowait(frame) def close(self) -> None: if self._closed: return self._closed = True + if self._rx_task is not None and self._rx_task is not asyncio.current_task(): + self._rx_task.cancel() + self._rx_task = None + # Install a terminal sentinel so a parked reader wakes promptly. A frame decoded before the + # cancellation lands may still queue behind it; harmless, as _raise_if_closed() gates later reads. + while not self._rx_queue.empty(): + self._rx_queue.get_nowait() + self._rx_queue.put_nowait(self._closed_error()) if self._tx_task is not None: self._tx_task.cancel() self._tx_task = None self._tx.abort_all(self._closed_error) + # Deregister before closing: cancel() is deferred but close() is immediate, so the callbacks would + # otherwise be torn down against a closed -- possibly recycled -- fd. + try: + loop = asyncio.get_running_loop() + except RuntimeError: # Closed outside a running loop; nothing can be registered either. + pass + else: + fd = self._sock.fileno() + if fd >= 0: + try: + loop.remove_reader(fd) + loop.remove_writer(fd) + except NotImplementedError: + # ProactorEventLoop implements neither; SocketCAN is Linux-only but tests run on any loop. + pass self._sock.close() def __repr__(self) -> str: @@ -181,19 +224,23 @@ def _is_transient_tx_error(ex: OSError) -> bool: return ex.errno in _TRANSIENT_TX_ERRNO def _encode(self, identifier: int, data: bytes) -> bytes: - if len(data) > 8: - if not self._fd: - raise ValueError( - f"SocketCAN interface {self._name} cannot send a {len(data)}-byte frame on Classic CAN" - ) + # The frame format follows the interface, not the payload length: every frame on an FD interface is + # an FD frame, as in the reference (cy_can_socketcan picks the FD/Classic vtable from the netdev MTU). + # + # REFERENCE PARITY: BRS is set on every FD frame, whereas the reference emits `.flags = CANFD_FDF` + # alone (cy_can_socketcan.c). Without BRS the data phase runs at the arbitration bit rate, + # forfeiting the throughput that is the point of FD. BRS does not apply to Classic CAN. + if self._fd: return _CANFD_FRAME_STRUCT.pack( socket.CAN_EFF_FLAG | (identifier & socket.CAN_EFF_MASK), len(data), - _CANFD_FDF, + _CANFD_FDF | _CANFD_BRS, 0, 0, data.ljust(64, b"\x00"), ) + if len(data) > 8: + raise ValueError(f"SocketCAN interface {self._name} cannot send a {len(data)}-byte frame on Classic CAN") return _CAN_FRAME_STRUCT.pack( socket.CAN_EFF_FLAG | (identifier & socket.CAN_EFF_MASK), len(data), diff --git a/src/pycyphal2/udp.py b/src/pycyphal2/udp.py index ff6088d76..6055d7018 100644 --- a/src/pycyphal2/udp.py +++ b/src/pycyphal2/udp.py @@ -52,10 +52,20 @@ IPv4_SUBJECT_ID_MAX = 0x7FFFFF TRANSFER_ID_MASK = (1 << 48) - 1 _MULTICAST_TTL = 16 +# Linux uapi in.h value; older CPython does not expose it in the socket module, so fall back to the literal. +_IP_MULTICAST_ALL_LINUX = getattr(socket, "IP_MULTICAST_ALL", 49) _SIOCGIFMTU = 0x8921 _CYPHAL_OVERHEAD_MAX = 100 _CYPHAL_MTU_LINK_MIN = 576 _RX_SESSION_LIFETIME_NS = round(30.0 * 1e9) +_HOUSEKEEPING_PERIOD = 1.0 # RX-session retirement cadence, independent of new traffic. +# Bounds the learned reverse-route cache against untrusted traffic with spoofable source UIDs, evicting the +# least-recently-seen entry. Deliberately not time-based: a route must not expire while its peer is briefly +# silent; under flooding an evicted route merely makes respond() fail cleanly with SendError. +_REMOTE_ENDPOINT_CAPACITY = 8192 +# Per-reassembler cap on concurrent sender sessions: TTL retirement bounds retention time but not +# cardinality, so a burst of unique (spoofable) source UIDs would otherwise grow unboundedly. +_RX_SESSION_CAPACITY = 4096 _RX_SLOT_COUNT = 8 _RX_TRANSFER_HISTORY_COUNT = 32 _SUBJECT_ID_MODULUS_MAX = IPv4_SUBJECT_ID_MAX - SUBJECT_ID_PINNED_MAX @@ -147,6 +157,16 @@ def _frame_is_valid(header: _FrameHeader, payload_chunk: bytes | memoryview) -> return (header.frame_payload_offset + len(payload_chunk)) <= header.transfer_payload_size +def _collect_send_errors(results: list[BaseException | None]) -> list[BaseException]: + """ + Per-interface failures out of a ``gather(..., return_exceptions=True)`` over ``send_on_iface``. + + Tests ``BaseException``, not ``Exception``: gather returns an individually cancelled child as a + ``CancelledError`` instance, which an ``Exception`` test would score as a successful delivery. + """ + return [r for r in results if isinstance(r, BaseException)] + + @dataclass(frozen=True) class _Fragment: offset: int @@ -236,7 +256,7 @@ def _find_left_neighbor(self, left: int) -> _Fragment | None: def _find_right_neighbor(self, right: int) -> _Fragment | None: candidate: _Fragment | None = None for frag in self.fragments: - if frag.offset < right: + if frag.offset <= right: # Inclusive: a fragment starting exactly at `right` is adjacent. candidate = frag else: break @@ -284,7 +304,10 @@ def is_transfer_ejected(self, transfer_id: int) -> bool: return transfer_id in self.history def initialize_history(self, transfer_id: int) -> None: - value = (transfer_id - 1) & TRANSFER_ID_MASK + # Seed wraps mod 2**64, not 2**48, mirroring the reference uint64 arithmetic: a 48-bit-masked seed + # for a first-seen transfer-ID of 0 would be 0xFFFF_FFFF_FFFF and falsely reject a genuine transfer + # bearing that ID. + value = (transfer_id - 1) & 0xFFFF_FFFF_FFFF_FFFF self.history = [value] * _RX_TRANSFER_HISTORY_COUNT self.history_current = 0 self.initialized = True @@ -322,6 +345,9 @@ def get_slot(self, timestamp_ns: int, header: _FrameHeader) -> tuple[int, _Trans class _RxReassembler: def __init__(self) -> None: + # LRU ordered most-recently-active FIRST, so the oldest is last: next(reversed(...)). + # This is the mirror image of _UDPTransportImpl._remote_endpoints (most-recent LAST); flipping + # either requires flipping every popitem()/reversed() that reads it. self._sessions: OrderedDict[int, _RxSession] = OrderedDict() def accept( @@ -346,6 +372,9 @@ def accept( self._sessions[header.sender_uid] = session session.last_animated_ns = timestamp_ns self._sessions.move_to_end(header.sender_uid, last=False) + if len(self._sessions) > _RX_SESSION_CAPACITY: + evicted_uid, _ = self._sessions.popitem(last=True) + _logger.debug("UDP reasm session cache full, evicted uid=%016x", evicted_uid) if not session.initialized: session.initialize_history(header.transfer_id) if session.is_transfer_ejected(header.transfer_id): @@ -398,6 +427,17 @@ def _retire_one_stale_session(self, timestamp_ns: int) -> None: self._sessions.pop(oldest_uid) _logger.debug("UDP reasm retire uid=%016x", oldest_uid) + def drop_stale_sessions(self, timestamp_ns: int) -> None: + """Retire stale sessions independent of traffic (the reference does this from a periodic poll). + Sessions are recency-ordered, so the scan stops at the first fresh one.""" + while self._sessions: + oldest_uid = next(reversed(self._sessions)) + if timestamp_ns >= (self._sessions[oldest_uid].last_animated_ns + _RX_SESSION_LIFETIME_NS): + self._sessions.pop(oldest_uid) + _logger.debug("UDP reasm retire uid=%016x", oldest_uid) + else: + break + def _make_subject_endpoint(subject_id: int) -> tuple[str, int]: """Return (multicast_ip, port) for a given subject_id.""" @@ -469,21 +509,25 @@ async def __call__(self, deadline: Instant, priority: Priority, message: bytes | self._transfer_id += 1 _logger.debug("Subject tx start sid=%d tid=%d bytes=%d", self._subject_id, transfer_id, len(message)) - errors: list[Exception] = [] - success_count = 0 - for i, iface in enumerate(self._transport.interfaces): - mtu = iface.mtu_cyphal - frames = _segment_transfer(priority, transfer_id, self._transport.uid, message, mtu) - try: - for frame in frames: - await self._transport.async_sendto(self._transport.tx_socks[i], frame, (mcast_ip, port), deadline) - success_count += 1 - except (OSError, SendError) as e: - errors.append(e) + addr = (mcast_ip, port) + # Snapshot before the first await: a concurrent close() clearing the socket lists would otherwise + # desync the indices; a send racing close just hits a closed socket and aggregates as an error. + targets = list(zip(self._transport.interfaces, self._transport.tx_socks, self._transport.tx_locks, strict=True)) + coros = [] + for iface, sock, lock in targets: + frames = _segment_transfer(priority, transfer_id, self._transport.uid, message, iface.mtu_cyphal) + coros.append(self._transport.send_on_iface(sock, lock, frames, addr, deadline)) + # Concurrent so a congested interface cannot starve a healthy one of the shared deadline; frames + # still go out in order per interface. return_exceptions lets every send settle before aggregating. + results = await asyncio.gather(*coros, return_exceptions=True) + errors = _collect_send_errors(results) + success_count = len(results) - len(errors) if errors and success_count == 0: _logger.error("Send failed on all interfaces for subject %d", self._subject_id) - raise SendError("send failed on all interfaces") from ExceptionGroup( + # BaseExceptionGroup because `errors` may carry a CancelledError, which ExceptionGroup rejects; + # it degrades to an ExceptionGroup when every member is an Exception. + raise SendError("send failed on all interfaces") from BaseExceptionGroup( "send failed on all interfaces", errors ) if errors: @@ -556,6 +600,10 @@ def new( detected. You can also use ``UDPTransport.list_interfaces()`` for a semi-automatic approach. The UID is a globally unique 64-bit identifier of the local node. If not given, one will be generated randomly. + + Overriding ``subject_id_modulus`` only makes sense for a deliberately reduced subject-ID space; the + value must be at least 57203, prime, and congruent to 3 modulo 4, else :meth:`pycyphal2.Node.new` + rejects it with ``ValueError``. This constructor only enforces the transport-level range. """ if not interfaces: ifaces = UDPTransport.list_interfaces() @@ -626,11 +674,20 @@ def __init__(self, interfaces: Iterable[Interface], uid: int, subject_id_modulus raise ValueError("At least one network interface is required") self._tx_socks: list[socket.socket] = [] + # asyncio's selector loop allows only one writer callback per fd, so concurrent senders on one + # socket must be serialized; otherwise a displaced sock_sendto hangs until its deadline. + self._tx_locks: list[asyncio.Lock] = [] self._self_endpoints: set[tuple[str, int]] = set() - for iface in self._interfaces: - sock = self._create_tx_socket(iface) - self._tx_socks.append(sock) - self._self_endpoints.add(sock.getsockname()[:2]) + try: + for iface in self._interfaces: + sock = self._create_tx_socket(iface) + self._tx_socks.append(sock) + self._tx_locks.append(asyncio.Lock()) + self._self_endpoints.add(sock.getsockname()[:2]) + except BaseException: + for sock in self._tx_socks: # Roll back sockets created before the failure. + sock.close() + raise self._subject_handlers: dict[int, Callable[[TransportArrival], None]] = {} self._subject_writers: dict[int, _UDPSubjectWriter] = {} @@ -639,15 +696,25 @@ def __init__(self, interfaces: Iterable[Interface], uid: int, subject_id_modulus self._unicast_handler: Callable[[TransportArrival], None] | None = None self._unicast_reassembler = _RxReassembler() - self._remote_endpoints: dict[tuple[int, int], tuple[str, int]] = {} + self._remote_endpoints: OrderedDict[tuple[int, int], tuple[str, int]] = OrderedDict() self._next_unicast_transfer_id = int.from_bytes(os.urandom(6), "little") self._unicast_rx_tasks: list[asyncio.Task[None]] = [] self._mcast_rx_tasks: dict[tuple[int, int], asyncio.Task[None]] = {} - for i, sock in enumerate(self._tx_socks): - task = self._loop.create_task(self._unicast_rx_loop(sock, i)) - self._unicast_rx_tasks.append(task) + # A half-built transport is never returned to the caller and hence never close()d, so a failure + # here must not leak the TX sockets or orphan the RX tasks already spawned. + try: + for i, sock in enumerate(self._tx_socks): + task = self._loop.create_task(self._unicast_rx_loop(sock, i)) + self._unicast_rx_tasks.append(task) + self._housekeeping_task = self._loop.create_task(self._housekeeping_loop()) + except BaseException: + for task in self._unicast_rx_tasks: + task.cancel() + for sock in self._tx_socks: + sock.close() + raise _logger.info( "UDPTransport initialized: uid=0x%016x, interfaces=%s, modulus=%d", @@ -659,10 +726,14 @@ def __init__(self, interfaces: Iterable[Interface], uid: int, subject_id_modulus @staticmethod def _create_tx_socket(iface: Interface) -> socket.socket: sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP) - sock.setblocking(False) - sock.bind((str(iface.address), 0)) - sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, _MULTICAST_TTL) - sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_IF, socket.inet_aton(str(iface.address))) + try: + sock.setblocking(False) + sock.bind((str(iface.address), 0)) + sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, _MULTICAST_TTL) + sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_IF, socket.inet_aton(str(iface.address))) + except BaseException: + sock.close() # Do not leak the fd if bind/setsockopt fails. + raise _logger.info("TX socket created on %s, bound to port %d", iface.address, sock.getsockname()[1]) return sock @@ -670,17 +741,30 @@ def _create_tx_socket(iface: Interface) -> socket.socket: def _create_mcast_socket(subject_id: int, iface: Interface) -> socket.socket: mcast_ip, port = _make_subject_endpoint(subject_id) sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP) - sock.setblocking(False) - sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) - if hasattr(socket, "SO_REUSEPORT"): - sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) - # Bind to multicast group address on Linux; INADDR_ANY on Windows - if sys.platform == "win32": - sock.bind(("", port)) - else: - sock.bind((mcast_ip, port)) - mreq = socket.inet_aton(mcast_ip) + socket.inet_aton(str(iface.address)) - sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq) + try: + sock.setblocking(False) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + if hasattr(socket, "SO_REUSEPORT"): + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) + # Bind to multicast group address on Linux; INADDR_ANY on Windows + if sys.platform == "win32": + sock.bind(("", port)) + else: + sock.bind((mcast_ip, port)) + mreq = socket.inet_aton(mcast_ip) + socket.inet_aton(str(iface.address)) + sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq) + # REFERENCE PARITY: the reference filters received datagrams by ingress interface index via + # recvmsg+IP_PKTINFO (udp_wrapper.c). asyncio has no sock_recvmsg, so on Linux the same delivery + # set comes from IP_MULTICAST_ALL=0, which restricts this socket to its own (group, interface) + # membership instead of the default any-interface delivery that would mislearn reverse routes on + # multi-homed hosts. macOS/BSD scope delivery per membership natively; on Windows the socket binds + # INADDR_ANY and Winsock may deliver cross-interface traffic, so multi-homed Windows hosts should + # configure at most one transport interface per multicast-reachable network. + if sys.platform == "linux": + sock.setsockopt(socket.IPPROTO_IP, _IP_MULTICAST_ALL_LINUX, 0) + except BaseException: + sock.close() # Do not leak the fd if bind/join fails. + raise _logger.info("Multicast socket for subject %d on %s (%s:%d)", subject_id, iface.address, mcast_ip, port) return sock @@ -701,6 +785,40 @@ def interfaces(self) -> list[Interface]: def tx_socks(self) -> list[socket.socket]: return self._tx_socks + @property + def tx_locks(self) -> list[asyncio.Lock]: + return self._tx_locks + + async def send_on_iface( + self, + sock: socket.socket, + lock: asyncio.Lock, + frames: list[bytes], + addr: tuple[str, int], + deadline: Instant, + ) -> Exception | None: + """Send one transfer's frames on one interface under that socket's lock; returns the failure + instead of raising, so the caller can aggregate per-interface results.""" + # Bound the lock wait by the same absolute deadline, so a short-deadline sender queued behind a + # long-deadline holder fails on its own budget rather than waiting out the holder's. + remaining_ns = deadline.ns - Instant.now().ns + if remaining_ns <= 0: + return SendError("Deadline exceeded") + try: + await asyncio.wait_for(lock.acquire(), timeout=remaining_ns * 1e-9) + except asyncio.TimeoutError: + return SendError("Deadline exceeded waiting for socket lock") + try: + if self._closed: + return ClosedError("Transport closed") + for frame in frames: + await self.async_sendto(sock, frame, addr, deadline) + return None + except (OSError, SendError) as e: + return e + finally: + lock.release() + def __repr__(self) -> str: addrs = ", ".join(str(i.address) for i in self._interfaces) return f"UDPTransport(uid=0x{self._uid:016x}, interfaces=[{addrs}], modulus={self._subject_id_modulus_val})" @@ -717,8 +835,26 @@ def remove_subject_listener(self, subject_id: int, handler: Callable[[TransportA task.cancel() sock = self._mcast_socks.pop(key, None) if sock is not None: + self._deregister_socket(sock) # Before close(); see the note in close(). sock.close() + def _deregister_socket(self, sock: socket.socket) -> None: + """ + Drop any selector callbacks for this socket while its descriptor is still open. + + Cancelling the owning task only schedules the teardown, so otherwise the removal lands after the + fd is closed -- possibly hitting an unrelated socket that has recycled the number. + """ + fd = sock.fileno() + if fd < 0: + return + try: + self._loop.remove_reader(fd) + self._loop.remove_writer(fd) + except NotImplementedError: + # ProactorEventLoop uses overlapped I/O, not selector callbacks: nothing to deregister. + pass + def remove_subject_writer(self, subject_id: int, writer: _UDPSubjectWriter) -> None: if self._subject_writers.get(subject_id) is writer: self._subject_writers.pop(subject_id, None) @@ -741,12 +877,18 @@ def subject_listen(self, subject_id: int, handler: Callable[[TransportArrival], raise ValueError(f"Subject {subject_id} already has an active listener") _logger.info("Subscribing to subject %d", subject_id) self._subject_handlers[subject_id] = handler - for i, iface in enumerate(self._interfaces): - key = (subject_id, i) - sock = self._create_mcast_socket(subject_id, iface) - self._mcast_socks[key] = sock - task = self._loop.create_task(self._mcast_rx_loop(sock, subject_id, i)) - self._mcast_rx_tasks[key] = task + try: + for i, iface in enumerate(self._interfaces): + key = (subject_id, i) + sock = self._create_mcast_socket(subject_id, iface) + self._mcast_socks[key] = sock + task = self._loop.create_task(self._mcast_rx_loop(sock, subject_id, i)) + self._mcast_rx_tasks[key] = task + except BaseException: + # Roll back so a later subject_listen for this subject is not blocked by the duplicate-handler + # check above. + self.remove_subject_listener(subject_id, handler) + raise return _UDPSubjectListener(self, subject_id, handler) def subject_advertise(self, subject_id: int) -> SubjectWriter: @@ -768,26 +910,27 @@ async def unicast(self, deadline: Instant, priority: Priority, remote_id: int, m self._next_unicast_transfer_id += 1 _logger.debug("Unicast tx start rid=%016x tid=%d bytes=%d", remote_id, transfer_id, len(message)) - errors: list[Exception] = [] - success_count = 0 - for i, iface in enumerate(self._interfaces): + # Snapshot targets (only interfaces with a known endpoint) before the first await, as in the + # subject writer. + coros = [] + for i, (iface, sock, lock) in enumerate(zip(self._interfaces, self._tx_socks, self._tx_locks, strict=True)): ep = self._remote_endpoints.get((remote_id, i)) if ep is None: _logger.debug("Unicast tx skip rid=%016x iface=%d reason=no-endpoint", remote_id, i) continue frames = _segment_transfer(priority, transfer_id, self._uid, message, iface.mtu_cyphal) - try: - for frame in frames: - await self.async_sendto(self._tx_socks[i], frame, ep, deadline) - success_count += 1 - except (OSError, SendError) as e: - errors.append(e) + coros.append(self.send_on_iface(sock, lock, frames, ep, deadline)) - if success_count == 0: - if errors: - raise SendError("Unicast failed on all interfaces") from errors[0] + if not coros: _logger.warning("No endpoint known for remote_id=0x%016x", remote_id) raise SendError("No endpoint known for remote_id") + + results = await asyncio.gather(*coros, return_exceptions=True) + errors = _collect_send_errors(results) + success_count = len(results) - len(errors) + + if success_count == 0: + raise SendError("Unicast failed on all interfaces") from errors[0] if errors: # Redundant transport: delivery via at least one interface is a success. Warn but do not # raise, otherwise a delivered transfer would be reported as failed and retried (mirrors @@ -805,21 +948,32 @@ def close(self) -> None: return self._closed = True _logger.info("Closing UDPTransport uid=0x%016x", self._uid) + self._housekeeping_task.cancel() for task in self._unicast_rx_tasks: task.cancel() self._unicast_rx_tasks.clear() for task in self._mcast_rx_tasks.values(): task.cancel() self._mcast_rx_tasks.clear() + # Deregister before closing: the cancellations above land on a later loop iteration while close() + # takes effect now, so the callbacks would otherwise be torn down against a closed -- possibly + # recycled -- fd. A send already parked in sock_sendto still unblocks on its own deadline. + for sock in [*self._tx_socks, *self._mcast_socks.values()]: + self._deregister_socket(sock) for sock in self._tx_socks: sock.close() for sock in self._mcast_socks.values(): sock.close() self._mcast_socks.clear() self._tx_socks.clear() + self._tx_locks.clear() self._subject_handlers.clear() self._subject_writers.clear() self._reassemblers.clear() + # Release RX-side state so a closed transport retains no sessions, learned routes, or handler refs. + self._unicast_reassembler = _RxReassembler() + self._remote_endpoints.clear() + self._unicast_handler = None async def _mcast_rx_loop(self, sock: socket.socket, subject_id: int, iface_idx: int) -> None: try: @@ -856,9 +1010,33 @@ async def _unicast_rx_loop(self, sock: socket.socket, iface_idx: int) -> None: except asyncio.CancelledError: _logger.debug("Unicast rx cancelled iface=%d", iface_idx) + async def _housekeeping_loop(self) -> None: + """Retire stale reassembly sessions periodically (the reference does this from its poll), so a + silent remote's session is reclaimed instead of lingering long past its last frame.""" + try: + while not self._closed: + await asyncio.sleep(_HOUSEKEEPING_PERIOD) + try: + now_ns = Instant.now().ns + self._unicast_reassembler.drop_stale_sessions(now_ns) + for reassembler in list(self._reassemblers.values()): + reassembler.drop_stale_sessions(now_ns) + except Exception: + # Retirement is not traffic-driven: a faulty sweep must not kill the loop. + _logger.exception("Stale session sweep failed; continuing") + except asyncio.CancelledError: + pass + def _learn_remote_endpoint(self, remote_id: int, iface_idx: int, src_ip: str, src_port: int) -> None: - existing = self._remote_endpoints.get((remote_id, iface_idx)) - self._remote_endpoints[(remote_id, iface_idx)] = (src_ip, src_port) + key = (remote_id, iface_idx) + existing = self._remote_endpoints.get(key) + self._remote_endpoints[key] = (src_ip, src_port) + # Ordered most-recently-seen LAST, so eviction pops the FRONT. This is the mirror image of + # _RxReassembler._sessions -- see the note there before changing either. + self._remote_endpoints.move_to_end(key) + if len(self._remote_endpoints) > _REMOTE_ENDPOINT_CAPACITY: + evicted, _ = self._remote_endpoints.popitem(last=False) + _logger.debug("Remote endpoint cache full, evicted rid=%016x iface=%d", evicted[0], evicted[1]) if existing != (src_ip, src_port): _logger.info("Remote endpoint rid=%016x iface=%d ep=%s:%d", remote_id, iface_idx, src_ip, src_port) diff --git a/tests/can/test_media_slcan.py b/tests/can/test_media_slcan.py index 4578f4614..33915eb5b 100644 --- a/tests/can/test_media_slcan.py +++ b/tests/can/test_media_slcan.py @@ -53,9 +53,11 @@ def test_parse_classic_extended_frames() -> None: assert parser.feed(b"T000001232ABCD\r") == [Frame(id=0x123, data=b"\xab\xcd")] assert parser.feed(b"T000001232abCd\r") == [Frame(id=0x123, data=b"\xab\xcd")] - assert parser.feed(b"t1231AA\r") == [Frame(id=0x123, data=b"\xaa")] - assert parser.feed(b"t7FF1AA\r") == [Frame(id=0x7FF, data=b"\xaa")] - assert parser.feed(b"t7FF0\r") == [Frame(id=0x7FF, data=b"")] + # Standard-ID 't' frames are dropped: Frame has no IDE discriminator, so forwarding them would alias + # extended frames with small IDs. + assert parser.feed(b"t1231AA\r") == [] + assert parser.feed(b"t7FF1AA\r") == [] + assert parser.feed(b"t7FF0\r") == [] assert parser.feed(b"T000001") == [] assert parser.feed(b"230\r") == [Frame(id=0x123, data=b"")] assert parser.feed(b"x1BADC0DE201AB\r") == [Frame(id=0x1BADC0DE, data=b"\x01\xab")] @@ -68,7 +70,7 @@ def test_parse_ignores_optional_frame_suffix() -> None: assert parser.feed(b"T000001232ABCDL\r") == [Frame(id=0x123, data=b"\xab\xcd")] assert parser.feed(b"T000001232ABCD1234L\r") == [Frame(id=0x123, data=b"\xab\xcd")] assert parser.feed(b"T000001232ABCDzzzz\r") == [Frame(id=0x123, data=b"\xab\xcd")] - assert parser.feed(b"t1231AAL\r") == [Frame(id=0x123, data=b"\xaa")] + assert parser.feed(b"t1231AAL\r") == [] # Standard-ID frames are dropped regardless of suffix. assert parser.feed(b"T000001232ABCD1234\x03\r") == [Frame(id=0x123, data=b"\xab\xcd")] assert parser.feed(b"T10AE6EFF8000000FF000000A07071\r") == [ Frame(id=0x10AE6EFF, data=b"\x00\x00\x00\xff\x00\x00\x00\xa0"), diff --git a/tests/can/test_pythoncan.py b/tests/can/test_pythoncan.py index 67a620a6a..34421783e 100644 --- a/tests/can/test_pythoncan.py +++ b/tests/can/test_pythoncan.py @@ -1185,6 +1185,48 @@ async def test_unit_mixed_fd_and_classic_payloads() -> None: _close_all(a, b) +async def test_unit_fd_flags_follow_interface_mode() -> None: + """Every FD-interface frame carries is_fd and bitrate_switch regardless of payload length; a Classic + interface sets neither. Always-on BRS is the deliberate divergence noted in socketcan._encode.""" + a, b = _virtual_pair(fd=True) + sent: list[_can.Message] = [] + orig_send = a._bus.send + + def recording_send(msg: _can.Message, timeout: float | None = None) -> None: + sent.append(msg) + orig_send(msg, timeout) + + a._bus.send = recording_send # type: ignore[method-assign] + try: + a.enqueue(0x00030000, [memoryview(b"tiny")], Instant.now() + 2.0) + a.enqueue(0x00030001, [memoryview(bytes(range(32)))], Instant.now() + 2.0) + for _ in range(2): + await asyncio.wait_for(b.receive(), timeout=2.0) + assert len(sent) == 2 + assert all(m.is_fd for m in sent) + assert all(m.bitrate_switch for m in sent) # Including the 4-byte payload. + finally: + _close_all(a, b) + + c, d = _virtual_pair() + sent_classic: list[_can.Message] = [] + orig_send_c = c._bus.send + + def recording_send_c(msg: _can.Message, timeout: float | None = None) -> None: + sent_classic.append(msg) + orig_send_c(msg, timeout) + + c._bus.send = recording_send_c # type: ignore[method-assign] + try: + c.enqueue(0x00030002, [memoryview(b"tiny")], Instant.now() + 2.0) + await asyncio.wait_for(d.receive(), timeout=2.0) + assert len(sent_classic) == 1 + assert not sent_classic[0].is_fd + assert not sent_classic[0].bitrate_switch + finally: + _close_all(c, d) + + async def test_unit_enqueue_same_id_preserves_order() -> None: a, b = _virtual_pair() try: @@ -1254,15 +1296,36 @@ def flaky_send(msg, timeout=None): async def test_unit_rx_bus_error_propagates() -> None: + """An RX failure is recorded at its source and surfaces to the reader with the cause attached.""" mock_bus = MagicMock(spec=_can.BusABC) - mock_bus.recv.side_effect = OSError("hardware gone") + err = OSError("hardware gone") + mock_bus.recv.side_effect = err mock_bus.channel_info = "mock:err" itf = PythonCANInterface(mock_bus) - with pytest.raises(ClosedError, match="receive failed"): + with pytest.raises(ClosedError) as caught: await asyncio.wait_for(itf.receive(), timeout=2.0) + assert itf._failure is err # Recorded by _fail() in the RX thread's handoff, not by receive(). + assert caught.value.__cause__ is err itf.close() +async def test_unit_clean_close_is_not_recorded_as_a_failure() -> None: + """receive() used to feed EVERY sentinel back through _fail(), including the plain ClosedError from an + explicit close, so a clean shutdown got reported as 'receive failed'.""" + mock_bus = MagicMock(spec=_can.BusABC) + mock_bus.recv.return_value = None # Idle bus: the reader parks on the queue. + mock_bus.channel_info = "mock:cleanclose" + itf = PythonCANInterface(mock_bus) + receiver = asyncio.create_task(itf.receive()) + await asyncio.sleep(0.05) + assert not receiver.done() # Parked. + + itf.close() + with pytest.raises(ClosedError): + await asyncio.wait_for(receiver, timeout=2.0) + assert itf._failure is None + + async def test_unit_multiple_close_with_failure() -> None: mock_bus = MagicMock(spec=_can.BusABC) mock_bus.recv.side_effect = _can.CanError("fail") diff --git a/tests/can/test_reassembly.py b/tests/can/test_reassembly.py index af62d51bd..482ff2a5a 100644 --- a/tests/can/test_reassembly.py +++ b/tests/can/test_reassembly.py @@ -4,6 +4,32 @@ from pycyphal2.can._wire import TransferKind +def test_cleanup_drops_slot_free_session_at_transfer_id_timeout() -> None: + """A slot-free session dies at the 2 s transfer-ID timeout since last admission (reference: + canard_poll); a session holding a slot lives for 30 s.""" + endpoint = Endpoint(kind=TransferKind.MESSAGE_16, port_id=7, on_transfer=lambda *_: None) + stale = RxSession.new(0) + stale.last_admission_ts_ns = 0 + fresh = RxSession.new(0) + fresh.last_admission_ts_ns = 2_500_000_000 + endpoint.sessions[42] = stale + endpoint.sessions[43] = fresh + + Reassembler.cleanup_sessions([endpoint], 3_000_000_000) # Stale is 3 s idle, fresh only 0.5 s. + + assert 42 not in endpoint.sessions + assert 43 in endpoint.sessions + + # A live slot keeps the session alive despite admission staleness, up to 30 s. + occupied = RxSession.new(0) + occupied.last_admission_ts_ns = 0 + occupied.slots[0] = RxSlot(start_ts_ns=0, transfer_id=0, iface_index=0, expected_toggle=False) + endpoint.sessions[44] = occupied + Reassembler.cleanup_sessions([endpoint], 25_000_000_000) # Slot is 25 s old: below the 30 s purge. + assert 44 in endpoint.sessions + assert endpoint.sessions[44].slots[0] is not None + + def test_cleanup_drops_session_after_30_seconds() -> None: received: list[bytes] = [] endpoint = Endpoint( diff --git a/tests/can/test_socketcan_unit.py b/tests/can/test_socketcan_unit.py index cfd0e6edf..18b8e22ff 100644 --- a/tests/can/test_socketcan_unit.py +++ b/tests/can/test_socketcan_unit.py @@ -2,6 +2,7 @@ import asyncio import errno +import os from pathlib import Path import sys import types @@ -19,6 +20,8 @@ class _FakeRawSocket: def __init__(self) -> None: self.calls: list[tuple[object, ...]] = [] + # A real fd: a fabricated number would risk deregistering an unrelated fd in this process. + self._rfd, self._wfd = os.pipe() def setblocking(self, enabled: bool) -> None: self.calls.append(("setblocking", enabled)) @@ -29,8 +32,17 @@ def setsockopt(self, level: int, option: int, value: object) -> None: def bind(self, address: tuple[str]) -> None: self.calls.append(("bind", address)) + def fileno(self) -> int: + return self._rfd + def close(self) -> None: self.calls.append(("close",)) + for fd in (self._rfd, self._wfd): + try: + os.close(fd) + except OSError: + pass + self._rfd = self._wfd = -1 class _TaskStub: @@ -50,6 +62,16 @@ def __init__(self, *, recv: list[object] | None = None, send: list[object] | Non self.send = list(send or []) self.sent_frames: list[bytes] = [] self.created_tasks: list[object] = [] + # Records selector deregistration so a test can assert it happens before the fd is closed. + self.deregistered: list[tuple[str, int]] = [] + + def remove_reader(self, fd: int) -> bool: + self.deregistered.append(("reader", fd)) + return False + + def remove_writer(self, fd: int) -> bool: + self.deregistered.append(("writer", fd)) + return False def create_task(self, coro: object) -> _TaskStub: if hasattr(coro, "close"): @@ -169,6 +191,8 @@ def _make_iface( iface._failure = failure iface._tx = TxQueue() iface._tx_task = None + iface._rx_queue = asyncio.Queue() + iface._rx_task = None return iface @@ -252,31 +276,126 @@ async def test_enqueue_purge_and_close_paths(monkeypatch: pytest.MonkeyPatch) -> closed.purge() -async def test_receive_retries_after_decode_drop_and_raises_on_failure(monkeypatch: pytest.MonkeyPatch) -> None: +async def test_rx_loop_decodes_skips_and_drops_cleanly_on_cancel(monkeypatch: pytest.MonkeyPatch) -> None: fake_socket, _ = _make_socket_module() module = _load_socketcan_module(monkeypatch, socket_module=fake_socket) iface = _make_iface(module) good = module._CAN_FRAME_STRUCT.pack(fake_socket.CAN_EFF_FLAG | 0x123, 2, b"ab".ljust(8, b"\x00")) - loop = _FakeLoop(recv=[b"\x00", good]) + loop = _FakeLoop(recv=[b"\x00", good, asyncio.CancelledError()]) # Undecodable, good, then cancelled. monkeypatch.setattr(module.asyncio, "get_running_loop", lambda: loop) - frame = await iface.receive() + with pytest.raises(asyncio.CancelledError): + await iface._rx_loop() + frame = iface._rx_queue.get_nowait() # The undecodable frame was dropped, not queued. + assert isinstance(frame, TimestampedFrame) assert frame.id == 0x123 assert frame.data == b"ab" + assert iface._rx_queue.empty() - failing = _make_iface(module) - failing_loop = _FakeLoop(recv=[OSError("rx failed")]) - monkeypatch.setattr(module.asyncio, "get_running_loop", lambda: failing_loop) - with pytest.raises(ClosedError, match="receive failed"): - await failing.receive() - assert failing._closed is True - assert isinstance(failing._failure, OSError) - cancelled = _make_iface(module) - cancelled_loop = _FakeLoop(recv=[asyncio.CancelledError()]) - monkeypatch.setattr(module.asyncio, "get_running_loop", lambda: cancelled_loop) - with pytest.raises(asyncio.CancelledError): - await cancelled.receive() +async def test_rx_loop_failure_marks_interface_and_installs_sentinel(monkeypatch: pytest.MonkeyPatch) -> None: + fake_socket, _ = _make_socket_module() + module = _load_socketcan_module(monkeypatch, socket_module=fake_socket) + iface = _make_iface(module) + err = OSError("rx failed") + loop = _FakeLoop(recv=[err]) + monkeypatch.setattr(module.asyncio, "get_running_loop", lambda: loop) + + await iface._rx_loop() # _fail()s the interface, then returns rather than raising. + assert iface._closed is True + assert iface._failure is err + sentinel = iface._rx_queue.get_nowait() + assert isinstance(sentinel, ClosedError) + + +async def test_receive_raises_clean_close_without_recording_failure(monkeypatch: pytest.MonkeyPatch) -> None: + """An explicit close reaches receive() as a plain ClosedError; only a receive-side error is recorded + as an interface failure.""" + fake_socket, _ = _make_socket_module() + module = _load_socketcan_module(monkeypatch, socket_module=fake_socket) + iface = _make_iface(module) + iface._rx_task = _TaskStub() # Already spawned, so receive() only drains the queue. + iface._rx_queue.put_nowait(ClosedError("SocketCAN interface vcan0 closed")) + with pytest.raises(ClosedError): + await iface.receive() + assert iface._failure is None + + +async def test_close_wakes_parked_receiver(monkeypatch: pytest.MonkeyPatch) -> None: + """close() must wake a parked reader with a ClosedError sentinel; otherwise interface loss leaves the + transport's reader hung forever.""" + fake_socket, _ = _make_socket_module() + module = _load_socketcan_module(monkeypatch, socket_module=fake_socket) + iface = _make_iface(module) + iface._rx_task = asyncio.create_task(asyncio.sleep(100)) # Stand-in for the RX loop task. + + recv_task = asyncio.create_task(iface.receive()) + await asyncio.sleep(0) + assert not recv_task.done() # Parked on the empty queue. + + iface.close() + with pytest.raises(ClosedError): + await asyncio.wait_for(recv_task, timeout=1.0) + assert iface._rx_task is None + assert iface._failure is None # A clean close is not an interface failure. + + +async def test_close_deregisters_the_fd_before_closing_it(monkeypatch: pytest.MonkeyPatch) -> None: + """Task.cancel() is deferred while socket.close() is immediate, so leaving deregistration to the + cancelled reader tears down callbacks against a closed -- possibly already reused -- fd. close() must + deregister explicitly, first.""" + fake_socket, _ = _make_socket_module() + module = _load_socketcan_module(monkeypatch, socket_module=fake_socket) + iface = _make_iface(module) + loop = _FakeLoop() + monkeypatch.setattr(module.asyncio, "get_running_loop", lambda: loop) + + sock = iface._sock + fd = sock.fileno() + assert fd >= 0 + iface.close() + + assert loop.deregistered == [("reader", fd), ("writer", fd)] + # The fd must still have been open at deregistration time. + assert ("close",) in sock.calls + assert loop.deregistered, "deregistration must not be left to the deferred task cancellation" + + +async def test_close_tolerates_a_loop_without_reader_registration(monkeypatch: pytest.MonkeyPatch) -> None: + """ProactorEventLoop raises NotImplementedError from remove_reader/writer; close() must treat that + as "nothing to deregister". An unguarded call broke every Windows job.""" + fake_socket, _ = _make_socket_module() + module = _load_socketcan_module(monkeypatch, socket_module=fake_socket) + iface = _make_iface(module) + + class _ProactorishLoop(_FakeLoop): + def remove_reader(self, fd: int) -> bool: + raise NotImplementedError + + def remove_writer(self, fd: int) -> bool: + raise NotImplementedError + + monkeypatch.setattr(module.asyncio, "get_running_loop", lambda: _ProactorishLoop()) + + iface.close() # Must not raise. + assert iface._closed + assert ("close",) in iface._sock.calls # The socket was still closed. + + +async def test_fail_wakes_parked_receiver(monkeypatch: pytest.MonkeyPatch) -> None: + """A non-transient TX failure (_fail -> close) must likewise unblock a parked reader.""" + fake_socket, _ = _make_socket_module() + module = _load_socketcan_module(monkeypatch, socket_module=fake_socket) + iface = _make_iface(module) + iface._rx_task = asyncio.create_task(asyncio.sleep(100)) + + recv_task = asyncio.create_task(iface.receive()) + await asyncio.sleep(0) + + iface._fail(OSError("ENETDOWN")) + with pytest.raises(ClosedError): + await asyncio.wait_for(recv_task, timeout=1.0) + assert isinstance(iface._failure, OSError) def test_raise_if_closed_and_transient_error_helpers(monkeypatch: pytest.MonkeyPatch) -> None: @@ -311,6 +430,20 @@ def test_encode_and_decode_branches(monkeypatch: pytest.MonkeyPatch) -> None: encoded_fd = fd_iface._encode(456, b"012345678") assert len(encoded_fd) == module._FD_FRAME_SIZE + # Format follows the interface mode, not the payload length, as in the reference. + encoded_fd_short = fd_iface._encode(456, b"abc") + assert len(encoded_fd_short) == module._FD_FRAME_SIZE + + # Every FD frame must carry FDF and BRS, short payloads included. The literals are hardcoded from + # linux/can.h (CPython's socket module exposes neither); a getattr() fallback to 0 would silently + # emit flagless frames. + assert module._CANFD_FDF == 0x04 + assert module._CANFD_BRS == 0x01 + for encoded in (encoded_fd, encoded_fd_short): + _, _, flags, _, _, _ = module._CANFD_FRAME_STRUCT.unpack(encoded) + assert flags & module._CANFD_FDF + assert flags & module._CANFD_BRS + assert module.SocketCANInterface._decode(b"\x00") is None non_extended = module._CAN_FRAME_STRUCT.pack(0x123, 1, b"x".ljust(8, b"\x00")) diff --git a/tests/can/test_transport_internal.py b/tests/can/test_transport_internal.py index 42154c488..019fc194b 100644 --- a/tests/can/test_transport_internal.py +++ b/tests/can/test_transport_internal.py @@ -10,7 +10,7 @@ from pycyphal2._transport import SUBJECT_ID_MODULUS_16bit, TransportArrival from pycyphal2.can import CANTransport, TimestampedFrame from pycyphal2.can._transport import _CANTransportImpl, _PinnedSubjectState -from pycyphal2.can._wire import NODE_ID_ANONYMOUS, TransferKind +from pycyphal2.can._wire import NODE_ID_ANONYMOUS, TransferKind, parse_frame, serialize_transfer from tests.can._support import MockCANBus, MockCANInterface, wait_for @@ -99,9 +99,17 @@ async def test_writer_unicast_and_send_transfer_error_paths() -> None: with pytest.raises(ClosedError, match="CAN transport closed"): await writer2(Instant.now() + 1.0, Priority.NOMINAL, b"x") - live = CANTransport.new(MockCANInterface(bus, "if1")) - with pytest.raises(ValueError, match="Invalid remote node-ID"): - await live.unicast(Instant.now() + 1.0, Priority.NOMINAL, 0, b"x") + live_if = MockCANInterface(bus, "if1") + live = CANTransport.new(live_if) + # Node-ID 0 is a legal unicast destination; the wire encoding must carry destination 0. + await live.unicast(Instant.now() + 1.0, Priority.NOMINAL, 0, b"x") + uni_id, uni_frames, _ = live_if.enqueue_history[-1] + parsed = parse_frame(uni_id, uni_frames[0]) + assert parsed is not None + assert parsed.destination_id == 0 + for bad_remote in (-1, 128): + with pytest.raises(ValueError, match="Invalid remote node-ID"): + await live.unicast(Instant.now() + 1.0, Priority.NOMINAL, bad_remote, b"x") live_impl = cast(_CANTransportImpl, live) with pytest.raises(SendError, match="Deadline exceeded"): @@ -220,6 +228,35 @@ async def test_reader_loop_exit_paths() -> None: delayed.close() +async def test_reader_loop_survives_raising_handler() -> None: + """A raising RX handler must not kill the interface's reader loop; the frame is dropped and reception continues.""" + bus = MockCANBus() + pub_if = MockCANInterface(bus, "pub") + sub_if = MockCANInterface(bus, "sub") + sub = CANTransport.new(sub_if) + calls: list[bytes] = [] + + def raising_handler(arrival: TransportArrival) -> None: + calls.append(bytes(arrival.message)) + raise ValueError("simulated handler fault (e.g. malformed gossip name)") + + sub.subject_listen(7, raising_handler) + for tid in (3, 4): + frame_id, frames = serialize_transfer( + kind=TransferKind.MESSAGE_16, + priority=0, + port_id=7, + source_id=55, + payload=b"payload", + transfer_id=tid, + fd=False, + ) + pub_if.enqueue(frame_id, [memoryview(frames[0])], Instant.now() + 1.0) + await wait_for(lambda: len(calls) == 2) + assert sub.interfaces == [sub_if] # The interface must not be dropped by the handler fault. + sub.close() + + async def test_drop_interface_and_node_id_occupancy_edges(caplog: pytest.LogCaptureFixture) -> None: caplog.set_level(logging.DEBUG) bus = MockCANBus() diff --git a/tests/mock_transport.py b/tests/mock_transport.py index ce84b838e..ec655c886 100644 --- a/tests/mock_transport.py +++ b/tests/mock_transport.py @@ -72,6 +72,10 @@ def __init__(self, node_id: int = 0, modulus: int = DEFAULT_MODULUS, network: Mo self.unicast_log: list[tuple[int, bytes]] = [] self.closed = False self.fail_unicast = False + # Setup-path failure injection for transactional-rollback tests: a subject-ID in either set makes + # that acquisition raise. Never auto-cleared, so a test controls exactly which retries fail. + self.fail_subject_listen: set[int] = set() + self.fail_subject_advertise: set[int] = set() if network is not None: network.add_transport(self) @@ -86,6 +90,8 @@ def subject_id_modulus(self) -> int: def subject_listen(self, subject_id: int, handler: Callable[[TransportArrival], None]) -> Closable: if subject_id in self.subject_handlers: raise ValueError(f"Subject {subject_id} already has an active listener") + if subject_id in self.fail_subject_listen: + raise RuntimeError(f"Simulated subject_listen failure for {subject_id}") self.subject_handlers[subject_id] = handler self.subject_listener_creations[subject_id] = self.subject_listener_creations.get(subject_id, 0) + 1 return MockSubjectListener(self, subject_id, handler) @@ -93,6 +99,8 @@ def subject_listen(self, subject_id: int, handler: Callable[[TransportArrival], def subject_advertise(self, subject_id: int) -> MockSubjectWriter: if subject_id in self.writers: raise ValueError(f"Subject {subject_id} already has an active writer") + if subject_id in self.fail_subject_advertise: + raise RuntimeError(f"Simulated subject_advertise failure for {subject_id}") writer = MockSubjectWriter(self, subject_id) self.writers[subject_id] = writer self.subject_writer_creations[subject_id] = self.subject_writer_creations.get(subject_id, 0) + 1 diff --git a/tests/test_close_streams.py b/tests/test_close_streams.py new file mode 100644 index 000000000..53c45ee78 --- /dev/null +++ b/tests/test_close_streams.py @@ -0,0 +1,181 @@ +"""Regression tests for findings #21 and #22: node close and implicit-topic GC must not orphan or hang an +outstanding response stream, and an open stream / in-flight reliable publish keeps its topic alive.""" + +from __future__ import annotations + +import asyncio + +import pytest + +import pycyphal2 +from pycyphal2._node import Association +from tests.mock_transport import MockNetwork, MockTransport +from tests.typing_helpers import new_node, request_stream + + +async def test_node_close_unblocks_parked_response_stream() -> None: + """Node.close() must cancel a library-owned request-publish task and stop a parked iteration, even + when the stream has an infinite response timeout (finding #21).""" + net = MockNetwork() + tr = MockTransport(node_id=1, network=net) + node = new_node(tr, home="n1") + pub = node.advertise("/rpc") + pub.ack_timeout = 0.05 + topic = node.topics_by_name["rpc"] + topic.associations[42] = Association(remote_id=42, last_seen=0.0) + + stream = await request_stream(pub, pycyphal2.Instant.now() + 1.0, float("inf"), b"request") + publish_task = stream._publish_task + anext_task = asyncio.create_task(stream.__anext__()) + await asyncio.sleep(0) + assert not anext_task.done() # Parked with no liveness timeout. + + node.close() + with pytest.raises(StopAsyncIteration): + await asyncio.wait_for(anext_task, timeout=1.0) + assert publish_task is not None + await asyncio.sleep(0) + assert publish_task.cancelled() or publish_task.done() + + +async def test_open_stream_keeps_topic_explicit_until_closed() -> None: + """An open response stream keeps its topic explicit after the publisher closes, so implicit GC cannot + destroy it; closing the stream lets the topic become implicit (finding #22).""" + net = MockNetwork() + tr = MockTransport(node_id=1, network=net) + node = new_node(tr, home="n1") + pub = node.advertise("/rpc") + pub.ack_timeout = 0.05 + topic = node.topics_by_name["rpc"] + topic.associations[42] = Association(remote_id=42, last_seen=0.0) + + stream = await request_stream(pub, pycyphal2.Instant.now() + 1.0, 1.0, b"request") + pub.close() + assert topic.is_implicit is False + + stream.close() # Its release re-syncs implicitness only on the next loop turn, hence the poll below. + for _ in range(50): + if topic.is_implicit: + break + await asyncio.sleep(0.001) + assert topic.is_implicit is True + + node.close() + + +async def test_implicit_gc_does_not_destroy_topic_with_open_stream(monkeypatch: pytest.MonkeyPatch) -> None: + """The implicit GC must not reap a topic while a response stream is open, only after it closes.""" + monkeypatch.setattr(pycyphal2._node, "IMPLICIT_TOPIC_TIMEOUT", 0.05) + net = MockNetwork() + tr = MockTransport(node_id=1, network=net) + node = new_node(tr, home="n1") + pub = node.advertise("/rpc") + pub.ack_timeout = 0.05 + topic = node.topics_by_name["rpc"] + topic.associations[42] = Association(remote_id=42, last_seen=0.0) + + stream = await request_stream(pub, pycyphal2.Instant.now() + 1.0, 1.0, b"request") + pub.close() + node.notify_implicit_gc() + await asyncio.sleep(0.15) + assert "rpc" in node.topics_by_name # Still alive: the open stream blocks GC. + + stream.close() + node.notify_implicit_gc() + for _ in range(50): + if "rpc" not in node.topics_by_name: + break + await asyncio.sleep(0.01) + assert "rpc" not in node.topics_by_name + + node.close() + + +async def test_zombie_stream_does_not_block_gc() -> None: + """A closed 'zombie' stream awaiting its dedup-cleanup timer must not keep the topic explicit, and + destroy_topic clears request_futures.""" + net = MockNetwork() + tr = MockTransport(node_id=1, network=net) + node = new_node(tr, home="n1") + pub = node.advertise("/rpc") + pub.ack_timeout = 0.05 + topic = node.topics_by_name["rpc"] + topic.associations[42] = Association(remote_id=42, last_seen=0.0) + + stream = await request_stream(pub, pycyphal2.Instant.now() + 1.0, 1.0, b"request") + # Populate a reliable remote so close() retains a zombie (scheduled cleanup rather than immediate). + from pycyphal2._publisher import ResponseRemoteState + + stream._reliable_remote_by_id[7] = ResponseRemoteState(seqno_top=0) + stream.close() + assert topic.request_futures # The zombie is retained pending its cleanup timer... + pub.close() + for _ in range(50): + if topic.is_implicit: + break + await asyncio.sleep(0.001) + assert topic.is_implicit is True # ...but a closed stream does not keep the topic explicit. + + node.destroy_topic("rpc") + assert topic.request_futures == {} + node.close() + + +async def test_request_rejects_nan_response_timeout() -> None: + net = MockNetwork() + tr = MockTransport(node_id=1, network=net) + node = new_node(tr, home="n1") + pub = node.advertise("/rpc") + with pytest.raises(ValueError, match="response_timeout"): + await pub.request(pycyphal2.Instant.now() + 1.0, float("nan"), b"request") + with pytest.raises(ValueError, match="response_timeout"): + await pub.request(pycyphal2.Instant.now() + 1.0, -1.0, b"request") + pub.close() + node.close() + + +async def test_pending_reliable_publish_keeps_topic_explicit() -> None: + """An in-flight reliable publish (tracked in publish_futures) keeps the topic explicit even with no + publisher, so implicit GC cannot destroy it mid-delivery (finding #22).""" + net = MockNetwork() + tr = MockTransport(node_id=1, network=net) + node = new_node(tr, home="n1") + pub = node.advertise("/rpc") + topic = node.topics_by_name["rpc"] + + from pycyphal2._node import PublishTracker + + topic.publish_futures[99] = PublishTracker(tag=99, ack_event=asyncio.Event()) + pub.close() + assert topic.is_implicit is False + + topic.publish_futures.clear() + topic.sync_implicit() + assert topic.is_implicit is True + + node.close() + + +async def test_close_does_not_resurrect_topic_via_disposed_stream_tail() -> None: + """A cancelled publish task's finally clause runs a loop iteration AFTER close() tore down the transport + and the gossip tasks, and it re-syncs topic implicitness; sync_topic_lifecycle must refuse to act on a + closed node, else it spawns an uncancellable gossip task and calls subject_listen on a dead transport.""" + net = MockNetwork() + tr = MockTransport(node_id=1, network=net) + node = new_node(tr, home="n1") + pub = node.advertise("/rpc") # Left OPEN, so pub_count keeps the topic explicit across close(). + pub.ack_timeout = 0.05 + topic = node.topics_by_name["rpc"] + topic.associations[42] = Association(remote_id=42, last_seen=0.0) + + stream = await request_stream(pub, pycyphal2.Instant.now() + 5.0, float("inf"), b"request") + assert stream._publish_task is not None + listener_creations = dict(tr.subject_listener_creations) + + node.close() + for _ in range(5): # Let the cancelled publish task's finally clause run to completion. + await asyncio.sleep(0) + + assert topic.gossip_task is None # No gossip task resurrected on a dead node. + assert tr.subject_listener_creations == listener_creations # No subject_listen on a closed transport. + assert not node._implicit_topics # close() cleared it and the tail did not re-populate it. diff --git a/tests/test_gossip.py b/tests/test_gossip.py index 8a2c58a80..ce35def5a 100644 --- a/tests/test_gossip.py +++ b/tests/test_gossip.py @@ -119,6 +119,54 @@ async def test_send_gossip_unicast(): node.close() +async def test_gossip_crafted_unicode_pin_name_does_not_raise(): + """A crafted gossip name like 'x#²' must be silently dropped, never raise, on the RX path.""" + net = MockNetwork() + tr = MockTransport(node_id=1, network=net) + node = new_node(tr, home="n1") + sub = node.subscribe("/sensor/>") + from pycyphal2._hash import rapidhash + + crafted = "x#²" # '²' passes str.isdigit() but is rejected by int() + name_bytes = crafted.encode("utf-8") + gossip_hdr = GossipHeader( + topic_log_age=5, + topic_hash=rapidhash(name_bytes), + topic_evictions=0, + name_len=len(name_bytes), + ) + arrival = TransportArrival( + timestamp=pycyphal2.Instant.now(), + priority=pycyphal2.Priority.NOMINAL, + remote_id=99, + message=gossip_hdr.serialize() + name_bytes, + ) + node.on_subject_arrival(node.broadcast_subject_id, arrival) # Must not raise. + assert crafted not in node.topics_by_name + + # The node must remain operational afterwards. + topic_name = "sensor/temp" + valid_hdr = GossipHeader( + topic_log_age=5, + topic_hash=rapidhash(topic_name), + topic_evictions=0, + name_len=len(topic_name), + ) + node.on_subject_arrival( + node.broadcast_subject_id, + TransportArrival( + timestamp=pycyphal2.Instant.now(), + priority=pycyphal2.Priority.NOMINAL, + remote_id=99, + message=valid_hdr.serialize() + topic_name.encode("utf-8"), + ), + ) + assert topic_name in node.topics_by_name + + sub.close() + node.close() + + async def test_gossip_implicit_topic_creation(): """A gossip whose name matches a pattern subscriber creates an implicit topic.""" net = MockNetwork() diff --git a/tests/test_housekeeping.py b/tests/test_housekeeping.py new file mode 100644 index 000000000..f25fec519 --- /dev/null +++ b/tests/test_housekeeping.py @@ -0,0 +1,140 @@ +"""Regression tests for finding #3: per-remote dedup and reordering state must be swept in aggregate on a +time basis, not only when the arriving remote sends again, so neither grows without bound.""" + +from __future__ import annotations + +import asyncio +import time + +import pytest + +import pycyphal2 +from pycyphal2._node import SESSION_LIFETIME, DedupState +from pycyphal2._subscriber import ReorderingState, SubscriberImpl +from tests.mock_transport import MockNetwork, MockTransport +from tests.typing_helpers import new_node + + +async def test_sweep_drops_stale_dedup_for_departed_remotes() -> None: + tr = MockTransport(node_id=1, network=MockNetwork()) + node = new_node(tr, home="n1") + sub = node.subscribe("/t") + topic = node.topics_by_name["t"] + + base = 1000.0 + topic.dedup[10] = DedupState(tag_frontier=1, last_active=base) + topic.dedup[11] = DedupState(tag_frontier=1, last_active=base) + topic.dedup[12] = DedupState(tag_frontier=1, last_active=base + 10_000.0) # Recently active. + + node.sweep_stale_states(base + SESSION_LIFETIME + 1.0) + assert 10 not in topic.dedup + assert 11 not in topic.dedup + assert 12 in topic.dedup # Recently active -> retained. + + sub.close() + node.close() + + +async def test_sweep_drops_stale_reordering_states() -> None: + tr = MockTransport(node_id=1, network=MockNetwork()) + node = new_node(tr, home="n1") + sub = node.subscribe("/t", reordering_window=0.2) + assert isinstance(sub, SubscriberImpl) + + base = 1000.0 + sub._reordering[(10, 0xAAAA)] = ReorderingState(last_active_at=base) + sub._reordering[(11, 0xBBBB)] = ReorderingState(last_active_at=base + 10_000.0) + + node.sweep_stale_states(base + SESSION_LIFETIME + 1.0) + assert (10, 0xAAAA) not in sub._reordering + assert (11, 0xBBBB) in sub._reordering # Recently active -> retained. + + sub.close() + node.close() + + +async def test_housekeeping_loop_sweeps_without_new_traffic(monkeypatch: pytest.MonkeyPatch) -> None: + """The background loop retires stale state on its own schedule even if the remote never sends again; + the sweep used to be triggered only by a new arrival.""" + monkeypatch.setattr(pycyphal2._node, "HOUSEKEEPING_PERIOD", 0.02) + tr = MockTransport(node_id=1, network=MockNetwork()) + node = new_node(tr, home="n1") + sub = node.subscribe("/t") + topic = node.topics_by_name["t"] + + topic.dedup[10] = DedupState(tag_frontier=1, last_active=time.monotonic() - SESSION_LIFETIME - 5.0) + for _ in range(100): + if 10 not in topic.dedup: + break + await asyncio.sleep(0.01) + assert 10 not in topic.dedup + + sub.close() + node.close() + + +async def test_housekeeping_loop_survives_a_raising_sweep(monkeypatch: pytest.MonkeyPatch) -> None: + """The sweep is the only bound on per-remote state growth, so the loop must outlive a faulty sweep: + catching only CancelledError let one stray exception kill the task silently (nothing retrieves its + result), leaving the node unbounded thereafter.""" + monkeypatch.setattr(pycyphal2._node, "HOUSEKEEPING_PERIOD", 0.02) + tr = MockTransport(node_id=1, network=MockNetwork()) + node = new_node(tr, home="n1") + sub = node.subscribe("/t") + topic = node.topics_by_name["t"] + + calls = 0 + real_sweep = node.sweep_stale_states + + def flaky_sweep(now: float) -> None: + nonlocal calls + calls += 1 + if calls == 1: + raise RuntimeError("simulated sweep fault") + real_sweep(now) + + monkeypatch.setattr(node, "sweep_stale_states", flaky_sweep) + + topic.dedup[10] = DedupState(tag_frontier=1, last_active=time.monotonic() - SESSION_LIFETIME - 5.0) + for _ in range(200): + if calls >= 2 and 10 not in topic.dedup: + break + await asyncio.sleep(0.01) + assert calls >= 2, "the loop died on the first faulty sweep" + assert 10 not in topic.dedup # Recovered and swept on a later tick. + assert not node._housekeeping_task.done() + + sub.close() + node.close() + + +async def test_implicit_gc_loop_survives_a_raising_retirement(monkeypatch: pytest.MonkeyPatch) -> None: + """A faulty retirement must not permanently disable implicit-topic GC. An expired topic yields a zero + delay, so the recovery path must also back off rather than spin at full speed.""" + monkeypatch.setattr(pycyphal2._node, "HOUSEKEEPING_PERIOD", 0.02) + monkeypatch.setattr(pycyphal2._node, "IMPLICIT_TOPIC_TIMEOUT", 0.02) + tr = MockTransport(node_id=1, network=MockNetwork()) + node = new_node(tr, home="n1") + + calls = 0 + real_retire = node._retire_one_expired_implicit_topic + + def flaky_retire(now: float) -> bool: + nonlocal calls + calls += 1 + if calls == 1: + raise RuntimeError("simulated retirement fault") + return real_retire(now) + + monkeypatch.setattr(node, "_retire_one_expired_implicit_topic", flaky_retire) + + node.topic_ensure("gc-me", None) # Implicit: no publisher, no subscriber. + for _ in range(200): + if calls >= 2 and "gc-me" not in node.topics_by_name: + break + await asyncio.sleep(0.01) + assert calls >= 2, "the loop died on the first faulty retirement" + assert "gc-me" not in node.topics_by_name # Recovered and retired the topic. + assert not node._gc_task.done() + + node.close() diff --git a/tests/test_names.py b/tests/test_names.py index c03731044..d9db10298 100644 --- a/tests/test_names.py +++ b/tests/test_names.py @@ -5,6 +5,7 @@ from pycyphal2 import SUBJECT_ID_PINNED_MAX from pycyphal2._node import ( TOPIC_NAME_MAX, + _is_valid_wire_name, _name_consume_pin_suffix, _name_normalize, match_pattern, @@ -78,6 +79,47 @@ def test_pin_non_digit_after_hash() -> None: assert _name_consume_pin_suffix("foo#abc") == ("foo#abc", None) +def test_pin_unicode_digit_not_parsed() -> None: + # str.isdigit() is True for chars int() rejects, e.g. '²' (U+00B2) or '②' (U+2461); must not raise. + assert _name_consume_pin_suffix("foo#²") == ("foo#²", None) + assert _name_consume_pin_suffix("foo#1²") == ("foo#1²", None) + assert _name_consume_pin_suffix("foo#²1") == ("foo#²1", None) + assert _name_consume_pin_suffix("x#②") == ("x#②", None) + + +def test_wire_name_unicode_digit_pin_rejected() -> None: + # A crafted gossip name like 'x#²' must be classified invalid without raising. + assert not _is_valid_wire_name("x#²") + assert not _is_valid_wire_name("x#②") + + +def test_resolve_embedded_token_is_verbatim() -> None: + # Only a whole segment '*' or '>' is a substitution token (reference: wkv_has_substitution_tokens); + # embedded in a longer segment they are literal characters. + resolved, _, verbatim = resolve_name("/sensor/temp*raw", "home", "ns") + assert resolved == "sensor/temp*raw" + assert verbatim + resolved, _, verbatim = resolve_name("/ab>cd", "home", "ns") + assert resolved == "ab>cd" + assert verbatim + + +def test_resolve_whole_segment_tokens_are_patterns() -> None: + assert resolve_name("/a/*/c", "home", "ns")[2] is False + assert resolve_name("/a/>", "home", "ns")[2] is False + assert resolve_name("/a/>/b", "home", "ns")[2] is False # Classified a pattern even off-terminal. + + +def test_wire_name_embedded_token_is_valid() -> None: + # A verbatim C topic like 'ab*cd' must be accepted from gossip for interop. + assert _is_valid_wire_name("ab*cd") + assert _is_valid_wire_name("x/y>z") + assert not _is_valid_wire_name("a/*/c") + assert not _is_valid_wire_name("a/>") + assert not _is_valid_wire_name("*") + assert not _is_valid_wire_name(">") + + def test_pin_hash_in_middle() -> None: # Pin is extracted from the rightmost '#' with a trailing digit run. assert _name_consume_pin_suffix("a#b#42") == ("a#b", 42) diff --git a/tests/test_parity.py b/tests/test_parity.py index f229dc4eb..0680ce47a 100644 --- a/tests/test_parity.py +++ b/tests/test_parity.py @@ -514,7 +514,9 @@ async def test_gossip_shard_formula(): async def test_broadcast_subject_id_formula(): """Broadcast subject-ID formula: broadcast_sid = (1 << (floor(log2(PINNED_MAX + modulus)) + 1)) - 1.""" - for modulus in [DEFAULT_MODULUS, 8378431, 131071, 65521]: + # All moduli satisfy the reference predicate (>= 57203, prime, ≡ 3 mod 4) yet span different log2 + # buckets of the formula. + for modulus in [DEFAULT_MODULUS, 8378431, 131071, 57203]: net = MockNetwork() tr = MockTransport(node_id=1, modulus=modulus, network=net) node = new_node(tr, home="n1") diff --git a/tests/test_reliable.py b/tests/test_reliable.py index 5ce3918d0..d35dc12ef 100644 --- a/tests/test_reliable.py +++ b/tests/test_reliable.py @@ -202,14 +202,16 @@ async def test_reliable_publish_retry_rebuilds_writer_and_header_after_reallocat async def test_gossip_reallocation_to_occupied_subject_preserves_writer(): net = MockNetwork() - tr = MockTransport(node_id=1, modulus=11, network=net) + # Smallest modulus satisfying the reference predicate (>= 57203, prime, ≡ 3 mod 4); the collision + # search below breaks on the first hit, so it runs in ~modulus iterations. + tr = MockTransport(node_id=1, modulus=57203, network=net) node = new_node(tr, home="n1") pub_a = node.advertise("/topic_a") topic_a = node.topics_by_name["topic_a"] target_sid = compute_subject_id(topic_a.hash, 1, tr.subject_id_modulus) colliding_name: str | None = None - for i in range(128): + for i in range(2_000_000): candidate = f"/topic_b_{i}" if compute_subject_id(rapidhash(candidate.removeprefix("/")), 0, tr.subject_id_modulus) == target_sid: colliding_name = candidate diff --git a/tests/test_topic.py b/tests/test_topic.py index ebc0e1494..ae85eb189 100644 --- a/tests/test_topic.py +++ b/tests/test_topic.py @@ -2,6 +2,8 @@ import time +import pytest + from pycyphal2 import SUBJECT_ID_PINNED_MAX from pycyphal2._node import left_wins from pycyphal2._hash import rapidhash @@ -9,6 +11,7 @@ EVICTIONS_PINNED_MIN, GossipScope, compute_subject_id, + is_valid_subject_id_modulus, match_pattern, resolve_name, ) @@ -41,7 +44,8 @@ def test_compute_subject_id_non_pinned_zero_evictions(): def test_compute_subject_id_non_pinned_with_evictions(): - """Non-pinned formula: offset + ((hash % modulus) + ((evictions % modulus)^2 % modulus)) % modulus.""" + """Non-pinned formula: offset + ((hash + evictions^2) mod 2^64) % modulus; the reduced form used for + the expectation below is equivalent absent 64-bit overflow.""" topic_hash = rapidhash("some/topic") for ev in (1, 2, 5, 100): sid = compute_subject_id(topic_hash, ev, DEFAULT_MODULUS) @@ -56,16 +60,17 @@ def test_compute_subject_id_non_pinned_with_evictions(): assert sid == expected -def test_compute_subject_id_non_pinned_does_not_wrap_uint64_sum(): +def test_compute_subject_id_wraps_uint64_sum(): + """hash + evictions² must wrap mod 2^64 as the reference uint64 arithmetic does. Evictions is an + untrusted gossip field, so the overflow is remotely constructible; big-int arithmetic here would put + Python and C nodes on different subject-IDs.""" topic_hash = (1 << 64) - 1 evictions = EVICTIONS_PINNED_MIN - 1 sid = compute_subject_id(topic_hash, evictions, DEFAULT_MODULUS) uint64_wrapping = ( SUBJECT_ID_PINNED_MAX + 1 + (((topic_hash + (evictions * evictions)) & ((1 << 64) - 1)) % DEFAULT_MODULUS) ) - assert sid == 49564 - assert uint64_wrapping == 74897 - assert sid != uint64_wrapping + assert sid == uint64_wrapping == 74897 def test_compute_subject_id_evictions_changes_sid(): @@ -92,6 +97,33 @@ def test_compute_subject_id_just_below_pinned(): assert sid == expected +def test_is_valid_subject_id_modulus_predicate(): + """Mirror of the reference predicate: >= 57203, prime, and ≡ 3 (mod 4).""" + for good in (57203, 122743, 8378431, 4294954663): + assert is_valid_subject_id_modulus(good) + assert not is_valid_subject_id_modulus(3) # Prime and ≡3 mod 4, but below the minimum. + assert not is_valid_subject_id_modulus(57202) # Below the minimum. + assert not is_valid_subject_id_modulus(57205) # ≡ 1 mod 4. + assert not is_valid_subject_id_modulus(57207) # ≡ 3 mod 4 but composite (3 × 19069). + assert not is_valid_subject_id_modulus(122744) # Even. + # Above uint32 is rejected without running a slow primality test on a huge untrusted value. + assert not is_valid_subject_id_modulus((1 << 32) + 3) + + +async def test_degenerate_subject_id_modulus_rejected(): + """A degenerate modulus must be rejected at node construction: the quadratic probe (hash + evictions²) + mod m would not cover the residue space, so topic_allocate's synchronous displacement loop would + hard-block the event loop.""" + for bad in (3, 57202, 57205, 57207, 122744): + tr = MockTransport(node_id=1, modulus=bad, network=MockNetwork()) + with pytest.raises(ValueError, match="subject_id_modulus"): + new_node(tr, home="n") + for good in (57203, 122743, 8378431): + tr = MockTransport(node_id=1, modulus=good, network=MockNetwork()) + node = new_node(tr, home="n") + node.close() + + async def test_advertise_creates_topic(): net = MockNetwork() tr = MockTransport(node_id=1, network=net) @@ -111,6 +143,22 @@ async def test_advertise_creates_topic(): node.close() +async def test_advertise_embedded_wildcard_char_is_verbatim(): + """'sensor/temp*raw' has no whole-segment substitution token, so it is a legal verbatim topic + (parity with the wkv classifier).""" + net = MockNetwork() + tr = MockTransport(node_id=1, network=net) + node = new_node(tr, home="n") + + pub = node.advertise("/sensor/temp*raw") + topic = node.topics_by_name.get("sensor/temp*raw") + assert topic is not None + assert topic.pub_count == 1 + + pub.close() + node.close() + + async def test_advertise_assigns_subject_id(): net = MockNetwork() tr = MockTransport(node_id=1, network=net) diff --git a/tests/test_transactional_setup.py b/tests/test_transactional_setup.py new file mode 100644 index 000000000..661effdf7 --- /dev/null +++ b/tests/test_transactional_setup.py @@ -0,0 +1,164 @@ +"""Regression tests for finding #11: setup paths are transactional (a mid-setup transport failure leaves +no unrepairable half-state), and subscribe-path listener failures follow the reference repair model +(logged, retried) rather than raising.""" + +from __future__ import annotations + +import math + +import pytest + +import pycyphal2 +from pycyphal2 import SUBJECT_ID_PINNED_MAX +from pycyphal2._hash import rapidhash +from pycyphal2._header import GossipHeader +from pycyphal2._transport import TransportArrival +from tests.mock_transport import DEFAULT_MODULUS, MockNetwork, MockTransport +from tests.typing_helpers import new_node + + +def _broadcast_sid(modulus: int = DEFAULT_MODULUS) -> int: + sid_max = SUBJECT_ID_PINNED_MAX + modulus + return (1 << (int(math.log2(sid_max)) + 1)) - 1 + + +async def test_node_init_rolls_back_broadcast_writer_on_listen_failure() -> None: + tr = MockTransport(node_id=1, network=MockNetwork()) + tr.fail_subject_listen.add(_broadcast_sid()) + with pytest.raises(RuntimeError, match="Simulated subject_listen"): + new_node(tr, home="n") + assert tr.writers == {} # Rolled back, so a retry sees no duplicate. + assert tr.subject_handlers == {} + + +async def test_node_init_rolls_back_broadcast_handles_on_unicast_listen_failure() -> None: + """unicast_listen is the LAST fallible acquisition in the constructor -- a plain assignment in both + built-in transports but not necessarily in a third-party one -- and nobody closes the transport when + the constructor raises, so an unguarded failure here strands the broadcast writer and listener.""" + tr = MockTransport(node_id=1, network=MockNetwork()) + + def failing_unicast_listen(_handler): # type: ignore[no-untyped-def] + raise RuntimeError("Simulated unicast_listen failure") + + tr.unicast_listen = failing_unicast_listen # type: ignore[method-assign] + with pytest.raises(RuntimeError, match="Simulated unicast_listen"): + new_node(tr, home="n") + assert tr.writers == {} + assert tr.subject_handlers == {} + + +async def test_advertise_rolls_back_pub_count_on_writer_failure() -> None: + # Learn the topic's subject-ID from a healthy node (it depends only on name + modulus). + healthy_tr = MockTransport(node_id=1, network=MockNetwork()) + healthy = new_node(healthy_tr, home="n") + pub = healthy.advertise("/topic_a") + topic_sid = healthy.topics_by_name["topic_a"].subject_id(healthy_tr.subject_id_modulus) + pub.close() + healthy.close() + + tr = MockTransport(node_id=1, network=MockNetwork()) + node = new_node(tr, home="n") + tr.fail_subject_advertise.add(topic_sid) + with pytest.raises(RuntimeError, match="Simulated subject_advertise"): + node.advertise("/topic_a") + + topic = node.topics_by_name["topic_a"] # Survives, but as an ordinary implicit topic. + assert topic.pub_count == 0 + assert topic.is_implicit + assert topic.pub_writer is None + + tr.fail_subject_advertise.clear() # Retry succeeds. + pub2 = node.advertise("/topic_a") + assert node.topics_by_name["topic_a"].pub_count == 1 + pub2.close() + node.close() + + +async def test_subscribe_listener_failure_is_repaired_not_raised() -> None: + healthy_tr = MockTransport(node_id=1, network=MockNetwork()) + healthy = new_node(healthy_tr, home="n") + sub_h = healthy.subscribe("/topic_v") + topic_sid = healthy.topics_by_name["topic_v"].subject_id(healthy_tr.subject_id_modulus) + sub_h.close() + healthy.close() + + tr = MockTransport(node_id=1, network=MockNetwork()) + node = new_node(tr, home="n") + tr.fail_subject_listen.add(topic_sid) + sub = node.subscribe("/topic_v") # Repair model: subscribe does not raise on listener failure. + topic = node.topics_by_name["topic_v"] + assert topic.sub_listener is None + creations_before = tr.subject_listener_creations.get(topic_sid, 0) + + tr.fail_subject_listen.clear() # The next sync opportunity repairs it. + topic.sync_listener() + assert topic.sub_listener is not None + assert tr.subject_listener_creations.get(topic_sid, 0) == creations_before + 1 + + sub.close() + node.close() + + +async def test_topic_ensure_rolls_back_on_gossip_shard_failure() -> None: + healthy_tr = MockTransport(node_id=1, network=MockNetwork()) + healthy = new_node(healthy_tr, home="n") + shard_sid = healthy.gossip_shard_subject_id(rapidhash("topic_s")) + healthy.close() + + tr = MockTransport(node_id=1, network=MockNetwork()) + node = new_node(tr, home="n") + tr.fail_subject_advertise.add(shard_sid) + with pytest.raises(RuntimeError, match="Simulated subject_advertise"): + node.advertise("/topic_s") + + assert "topic_s" not in node.topics_by_name # Both registries rolled back. + assert rapidhash("topic_s") not in node.topics_by_hash + + tr.fail_subject_advertise.clear() # Retry succeeds. + pub = node.advertise("/topic_s") + assert "topic_s" in node.topics_by_name + pub.close() + node.close() + + +async def test_gossip_driven_topic_creation_failure_never_raises() -> None: + net = MockNetwork() + tr = MockTransport(node_id=1, network=net) + node = new_node(tr, home="n") + sub = node.subscribe("/sensor/>") + + name = "sensor/temp" + shard_sid = node.gossip_shard_subject_id(rapidhash(name)) + hdr = GossipHeader(topic_log_age=5, topic_hash=rapidhash(name), topic_evictions=0, name_len=len(name)) + + def deliver() -> None: + node.on_subject_arrival( + node.broadcast_subject_id, + TransportArrival( + timestamp=pycyphal2.Instant.now(), + priority=pycyphal2.Priority.NOMINAL, + remote_id=99, + message=hdr.serialize() + name.encode("utf-8"), + ), + ) + + tr.fail_subject_advertise.add(shard_sid) + deliver() # Wire-driven path must not raise even when transport setup fails. + assert name not in node.topics_by_name + + tr.fail_subject_advertise.clear() # A later gossip creates the topic. + deliver() + assert name in node.topics_by_name + + sub.close() + node.close() + + +async def test_subscribe_bad_reordering_window_leaves_no_state() -> None: + tr = MockTransport(node_id=1, network=MockNetwork()) + node = new_node(tr, home="n") + with pytest.raises(ValueError, match="Reordering window"): + node.subscribe("/topic_r", reordering_window=-1.0) + assert "topic_r" not in node.sub_roots_verbatim + assert "topic_r" not in node.topics_by_name + node.close() diff --git a/tests/test_udp.py b/tests/test_udp.py index fb1401f97..f6f8ddade 100644 --- a/tests/test_udp.py +++ b/tests/test_udp.py @@ -1,8 +1,11 @@ from __future__ import annotations import asyncio +import errno import os +import socket import struct +import sys from ipaddress import IPv4Address from unittest.mock import patch @@ -31,9 +34,15 @@ Interface, UDPTransport, _FrameHeader, + _IP_MULTICAST_ALL_LINUX, + _REMOTE_ENDPOINT_CAPACITY, + _RX_SESSION_CAPACITY, + _RX_SESSION_LIFETIME_NS, _RxReassembler, + _RxSession, _SUBJECT_ID_MODULUS_MAX, _TransferSlot, + _collect_send_errors, _header_deserialize, _header_serialize, _make_subject_endpoint, @@ -265,6 +274,18 @@ def test_transfer_id_dedup(self): result2 = reasm.accept(frame_pairs[0][0], frame_pairs[0][1]) assert result2 is None # Dedup + def test_history_seed_sentinel_not_matchable_by_wire_id(self): + """A first-seen transfer-ID of 0 seeds the dedup history with (0 - 1) wrapped to 2^64-1, unmatchable + by any 48-bit wire ID (the reference keeps the unmasked uint64); a 48-bit-masked seed would equal + the valid wire value 0xFFFF_FFFF_FFFF and falsely reject it.""" + reasm = _RxReassembler() + first = self._make_frames(b"first", mtu=1400, transfer_id=0) + assert reasm.accept(first[0][0], first[0][1]) is not None + genuine = self._make_frames(b"genuine", mtu=1400, transfer_id=TRANSFER_ID_MASK) + result = reasm.accept(genuine[0][0], genuine[0][1]) + assert result is not None + assert result.payload == b"genuine" + def test_crc_mismatch_first_frame(self): payload = b"corrupt me" reasm = _RxReassembler() @@ -413,6 +434,32 @@ def test_ninth_concurrent_transfer_sacrifices_oldest_slot(self): slot_transfer_ids = {slot.transfer_id for slot in session.slots if slot is not None} assert slot_transfer_ids == set(range(2, 10)) + def test_drop_stale_sessions_retires_idle_without_new_traffic(self): + """Sessions past their lifetime are retired independently of new frames, so a silent remote is + reclaimed on the periodic tick (finding #3).""" + reasm = _RxReassembler() + old = self._make_frames(b"old", mtu=1400, sender_uid=100, transfer_id=1) + reasm.accept(old[0][0], old[0][1], timestamp_ns=0) + recent = self._make_frames(b"recent", mtu=1400, sender_uid=200, transfer_id=1) + # A timestamp within the lifetime so the built-in per-accept retire does not fire yet. + reasm.accept(recent[0][0], recent[0][1], timestamp_ns=1000) + assert set(reasm._sessions) == {100, 200} + + reasm.drop_stale_sessions(_RX_SESSION_LIFETIME_NS + 1) + assert 100 not in reasm._sessions # Idle past its lifetime -> retired. + assert 200 in reasm._sessions # Still within its lifetime -> retained. + + def test_sessions_bounded_by_lru_capacity(self): + """Session count is capacity-bounded: a burst of unique source UIDs within the lifetime window + cannot grow the map without bound (finding #3).""" + reasm = _RxReassembler() + for uid in range(_RX_SESSION_CAPACITY + 20): + frames = self._make_frames(b"x", mtu=1400, sender_uid=uid, transfer_id=1) + reasm.accept(frames[0][0], frames[0][1], timestamp_ns=uid) # Distinct, monotonically-advancing ts. + assert len(reasm._sessions) == _RX_SESSION_CAPACITY + assert 0 not in reasm._sessions # Least-recently-active evicted. + assert (_RX_SESSION_CAPACITY + 19) in reasm._sessions # Most recent retained. + def test_duplicate_history_window_is_32(self): reasm = _RxReassembler() for transfer_id in range(1, 34): @@ -466,6 +513,29 @@ def test_bridge_fragment_evicts_victim(self): assert slot._accept_fragment(2, b"XXXXXX") assert [(frag.offset, frag.data) for frag in slot.fragments] == [(0, b"AAAA"), (2, b"XXXXXX"), (6, b"CCCC")] + def test_conflicting_overlap_evicted_via_inclusive_right_neighbor(self): + """cavl2_predecessor is an inclusive floor, so a fragment starting exactly at the new one's end is + its right neighbor and the conflicting fragment between them is evicted; a strict '<' lookup would + keep the stale fragment and fail the transfer CRC.""" + payload = b"ABCDEFGH" + + def hdr(offset: int, crc: int = 0) -> _FrameHeader: + return _FrameHeader( + priority=4, + transfer_id=1, + sender_uid=1, + frame_payload_offset=offset, + transfer_payload_size=len(payload), + prefix_crc=crc, + ) + + slot = _TransferSlot.create(hdr(0), 0) + assert slot.update(0, hdr(2), b"XXXX") is None # Conflicting (corrupt/injected) overlap. + assert slot.update(1, hdr(3, crc32c_full(payload)), b"DEFGH") is None + result = slot.update(2, hdr(0), b"ABC") + assert [(frag.offset, bytes(frag.data)) for frag in slot.fragments] == [(0, b"ABC"), (3, b"DEFGH")] + assert result == payload + def test_furthest_reaching_crc_is_used(self): payload = b"abcdef" slot = _TransferSlot.create( @@ -903,6 +973,16 @@ async def test_operations_after_close_fail(self): with pytest.raises(SendError): await writer(Instant.now() + 1.0, Priority.NOMINAL, b"should fail") + @pytest.mark.skipif(sys.platform != "linux", reason="IP_MULTICAST_ALL is a Linux-only socket option") + def test_mcast_socket_disables_cross_interface_delivery(self, loopback_iface): + """IP_MULTICAST_ALL=0 confines the RX socket to its own (group, interface) membership -- the kernel + equivalent of the reference's recvmsg+IP_PKTINFO ingress-interface filter (udp_wrapper.c).""" + sock = _UDPTransportImpl._create_mcast_socket(5, loopback_iface) + try: + assert sock.getsockopt(socket.IPPROTO_IP, _IP_MULTICAST_ALL_LINUX) == 0 + finally: + sock.close() + @pytest.mark.asyncio async def test_subject_id_modulus(self, loopback_iface): t = UDPTransport.new(interfaces=[loopback_iface], subject_id_modulus=_SUBJECT_ID_MODULUS_MAX) @@ -928,6 +1008,26 @@ async def test_malformed_frame_does_not_learn_endpoint(self): finally: t.close() + @pytest.mark.asyncio + async def test_raising_subject_handler_does_not_escape(self): + """A raising subject handler is contained by the per-datagram fault boundary; reception continues.""" + t = UDPTransport.new_loopback() + assert isinstance(t, _UDPTransportImpl) + try: + calls: list[int] = [] + + def handler(arrival: TransportArrival) -> None: + calls.append(arrival.remote_id) + raise ValueError("simulated handler fault (e.g. malformed gossip name)") + + t._subject_handlers[55] = handler + for tid in (1, 2): + frame = _segment_transfer(4, tid, 0xAA, b"hello", mtu=1400)[0] + t._process_subject_datagram(frame, "10.0.0.1", 9000, 55, 0, Instant(ns=tid)) # Must not raise. + assert calls == [0xAA, 0xAA] + finally: + t.close() + @pytest.mark.asyncio async def test_transfer_failure_still_learns_endpoint(self): t = UDPTransport.new_loopback() @@ -1125,6 +1225,298 @@ async def all_fail(sock, data, addr, deadline): # type: ignore[no-untyped-def] pub.close() +def test_collect_send_errors_counts_cancellation_as_a_failure() -> None: + """gather(return_exceptions=True) reports a cancelled child as a CancelledError instance (a + BaseException), so the old isinstance(r, Exception) test scored a cancelled interface as a delivery.""" + oserr = OSError("down") + assert _collect_send_errors([None, None]) == [] + assert _collect_send_errors([None, oserr]) == [oserr] + cancelled = asyncio.CancelledError() + assert _collect_send_errors([None, cancelled]) == [cancelled] + assert len(_collect_send_errors([cancelled, asyncio.CancelledError()])) == 2 + + +async def test_send_cancelled_on_every_interface_is_not_reported_as_success() -> None: + """An all-cancelled send used to return normally, reporting a transfer that never hit the wire.""" + iface = Interface(address=IPv4Address("127.0.0.1"), mtu_link=1500) + pub = UDPTransport.new(interfaces=[iface, iface]) + assert isinstance(pub, _UDPTransportImpl) + try: + + async def cancelled_send(sock, lock, frames, addr, deadline): # type: ignore[no-untyped-def] + raise asyncio.CancelledError + + with patch.object(pub, "send_on_iface", cancelled_send): + writer = pub.subject_advertise(10) + with pytest.raises(SendError): + await writer(Instant.now() + 2.0, Priority.NOMINAL, b"nope") + + # Redundancy still holds: one cancelled interface alongside one delivery is a success. + async def cancel_first(sock, lock, frames, addr, deadline): # type: ignore[no-untyped-def] + if sock is pub.tx_socks[0]: + raise asyncio.CancelledError + return None + + with patch.object(pub, "send_on_iface", cancel_first): + writer_partial = pub.subject_advertise(11) + await writer_partial(Instant.now() + 2.0, Priority.NOMINAL, b"ok") + finally: + pub.close() + + +@pytest.mark.asyncio +async def test_redundant_interfaces_send_concurrently() -> None: + """A congested interface must not starve a healthy one of the shared deadline: sends run concurrently, + so a transfer's wall-clock is ~max(per-iface), not the sum (finding #6).""" + iface = Interface(address=IPv4Address("127.0.0.1"), mtu_link=1500) + pub = UDPTransport.new(interfaces=[iface, iface, iface]) + assert isinstance(pub, _UDPTransportImpl) + try: + + async def slow_sendto(sock, data, addr, deadline): # type: ignore[no-untyped-def] + await asyncio.sleep(0.2) + + with patch.object(pub, "async_sendto", slow_sendto): + writer = pub.subject_advertise(10) + start = asyncio.get_running_loop().time() + await writer(Instant.now() + 5.0, Priority.NOMINAL, b"x") + elapsed = asyncio.get_running_loop().time() - start + assert elapsed < 0.4, f"serial send would take >=0.6s across 3 interfaces, took {elapsed:.3f}s" + finally: + pub.close() + + +@pytest.mark.asyncio +async def test_concurrent_sends_on_shared_socket_are_serialized() -> None: + """Two transfers racing on the same per-interface socket must be serialized by its lock, so they + never overlap inside sock_sendto (the selector loop allows only one writer per fd) (finding #7).""" + iface = Interface(address=IPv4Address("127.0.0.1"), mtu_link=1500) + pub = UDPTransport.new(interfaces=[iface]) + assert isinstance(pub, _UDPTransportImpl) + try: + in_flight: set[int] = set() + overlap_detected = False + + async def tracking_sendto(sock, data, addr, deadline): # type: ignore[no-untyped-def] + nonlocal overlap_detected + fd = sock.fileno() + if fd in in_flight: + overlap_detected = True + in_flight.add(fd) + await asyncio.sleep(0.02) + in_flight.discard(fd) + + with patch.object(pub, "async_sendto", tracking_sendto): + w1 = pub.subject_advertise(10) + w2 = pub.subject_advertise(11) + await asyncio.gather( + w1(Instant.now() + 5.0, Priority.NOMINAL, b"a"), + w2(Instant.now() + 5.0, Priority.NOMINAL, b"b"), + w1(Instant.now() + 5.0, Priority.NOMINAL, b"c"), + ) + assert not overlap_detected + finally: + pub.close() + + +@pytest.mark.asyncio +async def test_short_deadline_send_fails_on_own_budget_behind_long_holder() -> None: + """A short-deadline sender queued behind a long-deadline holder of the same socket lock fails on its + own deadline rather than waiting out the holder: lock acquisition is deadline-bounded.""" + iface = Interface(address=IPv4Address("127.0.0.1"), mtu_link=1500) + pub = UDPTransport.new(interfaces=[iface]) + assert isinstance(pub, _UDPTransportImpl) + holder_entered = asyncio.Event() + holder_release = asyncio.Event() + + async def slow_sendto(sock, data, addr, deadline): # type: ignore[no-untyped-def] + holder_entered.set() + await holder_release.wait() + + with patch.object(pub, "async_sendto", slow_sendto): + holder = pub.subject_advertise(10) + waiter = pub.subject_advertise(11) # Same interface -> same socket lock. + holder_task = asyncio.create_task(holder(Instant.now() + 100.0, Priority.NOMINAL, b"hold")) + await holder_entered.wait() + with pytest.raises(SendError): + await waiter(Instant.now() + 0.15, Priority.NOMINAL, b"wait") + holder_release.set() + await holder_task + + # The lock-acquisition timeout must not have leaked the socket lock. + assert not pub._tx_locks[0].locked() + + async def ok_sendto(sock, data, addr, deadline): # type: ignore[no-untyped-def] + pass # Avoid a real multicast send (unroutable on Windows loopback); still re-acquires the lock. + + with patch.object(pub, "async_sendto", ok_sendto): + await waiter(Instant.now() + 2.0, Priority.NOMINAL, b"after") + pub.close() + + +@pytest.mark.asyncio +async def test_close_during_send_raises_send_error_not_index_error() -> None: + """A send suspended when close() empties the socket lists must surface a clean SendError, never an + IndexError (finding #9); the resumed send fails on the by-then-closed fd, as a real EBADF would.""" + iface = Interface(address=IPv4Address("127.0.0.1"), mtu_link=1500) + pub = UDPTransport.new(interfaces=[iface]) + assert isinstance(pub, _UDPTransportImpl) + started = asyncio.Event() + release = asyncio.Event() + + async def blocking_sendto(sock, data, addr, deadline): # type: ignore[no-untyped-def] + started.set() + await release.wait() + raise OSError(errno.EBADF, "Bad file descriptor") # What the real closed socket raises. + + with patch.object(pub, "async_sendto", blocking_sendto): + writer = pub.subject_advertise(10) + send_task = asyncio.create_task(writer(Instant.now() + 5.0, Priority.NOMINAL, b"payload")) + await started.wait() + pub.close() # Clears _tx_socks/_tx_locks; the snapshot keeps the parked send index-safe. + release.set() + with pytest.raises(SendError) as excinfo: + await send_task + cause = excinfo.value.__cause__ + causes = list(cause.exceptions) if isinstance(cause, BaseExceptionGroup) else [cause] + assert not any(isinstance(c, IndexError) for c in causes) + + +@pytest.mark.asyncio +async def test_subject_listen_rolls_back_partial_interface_setup() -> None: + """A per-interface socket failure mid-subject_listen rolls back the handler and every socket/task + created so far, so a retry is not blocked by the duplicate-listener check (finding #11).""" + iface = Interface(address=IPv4Address("127.0.0.1"), mtu_link=1500) + t = UDPTransport.new(interfaces=[iface, iface]) + assert isinstance(t, _UDPTransportImpl) + try: + real_create = t._create_mcast_socket + calls = {"n": 0} + + def flaky_create(subject_id, iface): # type: ignore[no-untyped-def] + calls["n"] += 1 + if calls["n"] == 2: # Fail on the second interface, after the first has been set up. + raise OSError("second interface bind failed") + return real_create(subject_id, iface) + + with patch.object(t, "_create_mcast_socket", flaky_create): + with pytest.raises(OSError): + t.subject_listen(42, lambda _a: None) + + assert 42 not in t._subject_handlers + assert not any(sid == 42 for (sid, _i) in t._mcast_socks) + assert not any(sid == 42 for (sid, _i) in t._mcast_rx_tasks) + + listener = t.subject_listen(42, lambda _a: None) # Retry succeeds. + listener.close() + finally: + t.close() + + +@pytest.mark.asyncio +async def test_remote_endpoints_bounded_by_lru_capacity() -> None: + """The learned reverse-route cache is LRU-bounded, so spoofable source UIDs cannot grow it without + bound (finding #3).""" + t = UDPTransport.new_loopback() + assert isinstance(t, _UDPTransportImpl) + try: + for uid in range(_REMOTE_ENDPOINT_CAPACITY + 50): + t._learn_remote_endpoint(uid, 0, "10.0.0.1", 9000) + assert len(t._remote_endpoints) == _REMOTE_ENDPOINT_CAPACITY + assert (0, 0) not in t._remote_endpoints # Least-recently-seen evicted. + assert (_REMOTE_ENDPOINT_CAPACITY + 49, 0) in t._remote_endpoints # Most recent retained. + finally: + t.close() + + +@pytest.mark.asyncio +async def test_close_releases_unicast_and_endpoint_state() -> None: + """close() must release the state that used to survive it: the unicast reassembler's sessions, the + learned endpoint cache, and the unicast handler reference (finding #3).""" + t = UDPTransport.new_loopback() + assert isinstance(t, _UDPTransportImpl) + t.unicast_listen(lambda _a: None) + t._learn_remote_endpoint(5, 0, "10.0.0.1", 9000) + t._unicast_reassembler._sessions[100] = _RxSession(last_animated_ns=0) + + t.close() + assert t._remote_endpoints == {} + assert t._unicast_handler is None + assert t._unicast_reassembler._sessions == {} + + +@pytest.mark.asyncio +async def test_housekeeping_loop_retires_stale_sessions(monkeypatch: pytest.MonkeyPatch) -> None: + """The periodic housekeeping loop retires stale reassembly sessions with no further traffic.""" + monkeypatch.setattr("pycyphal2.udp._HOUSEKEEPING_PERIOD", 0.02) + monkeypatch.setattr("pycyphal2.udp._RX_SESSION_LIFETIME_NS", 1) + t = UDPTransport.new_loopback() + assert isinstance(t, _UDPTransportImpl) + try: + t._unicast_reassembler._sessions[100] = _RxSession(last_animated_ns=0) + for _ in range(100): + if not t._unicast_reassembler._sessions: + break + await asyncio.sleep(0.01) + assert t._unicast_reassembler._sessions == {} + finally: + t.close() + + +async def test_housekeeping_loop_survives_a_raising_sweep(monkeypatch: pytest.MonkeyPatch) -> None: + """Session retirement is not traffic-driven, so the loop must outlive a faulty sweep: catching only + CancelledError let one stray exception kill the task silently, leaking sessions thereafter.""" + monkeypatch.setattr("pycyphal2.udp._HOUSEKEEPING_PERIOD", 0.02) + monkeypatch.setattr("pycyphal2.udp._RX_SESSION_LIFETIME_NS", 1) + t = UDPTransport.new_loopback() + assert isinstance(t, _UDPTransportImpl) + try: + calls = {"n": 0} + real_drop = _RxReassembler.drop_stale_sessions + + def flaky_drop(self, now_ns): # type: ignore[no-untyped-def] + calls["n"] += 1 + if calls["n"] == 1: + raise RuntimeError("simulated sweep fault") + real_drop(self, now_ns) + + with patch.object(_RxReassembler, "drop_stale_sessions", flaky_drop): + t._unicast_reassembler._sessions[100] = _RxSession(last_animated_ns=0) + for _ in range(200): + if calls["n"] >= 2 and not t._unicast_reassembler._sessions: + break + await asyncio.sleep(0.01) + assert calls["n"] >= 2, "the loop died on the first faulty sweep" + assert t._unicast_reassembler._sessions == {} + assert not t._housekeeping_task.done() + finally: + t.close() + + +@pytest.mark.asyncio +async def test_tx_socket_creation_failure_rolls_back_created_sockets() -> None: + """A TX-socket creation failure mid-construction rolls back the sockets created before it instead of + leaking their file descriptors (finding #11 completeness).""" + iface = Interface(address=IPv4Address("127.0.0.1"), mtu_link=1500) + real_create = _UDPTransportImpl._create_tx_socket + created: list[socket.socket] = [] + calls = {"n": 0} + + def flaky_create(iface_arg): # type: ignore[no-untyped-def] + calls["n"] += 1 + if calls["n"] == 2: # Fail the second interface, after the first socket exists. + raise OSError("tx socket bind failed") + sock = real_create(iface_arg) + created.append(sock) + return sock + + with patch.object(_UDPTransportImpl, "_create_tx_socket", staticmethod(flaky_create)): + with pytest.raises(OSError): + UDPTransport.new(interfaces=[iface, iface]) + assert len(created) == 1 + assert created[0].fileno() == -1 # The first socket was closed by the rollback (no fd leak). + + def test_interface_rejects_subminimum_mtu() -> None: """A link MTU below the Cyphal minimum is rejected at construction, not via a strippable assert.""" with pytest.raises(ValueError, match="mtu_link must be"):