feat(mld): MLDv2 state-change robustness retransmit train (R4 P4b)

Adds the RFC 3810 §6.1 robustness retransmission of unsolicited MLDv2
state-change Reports — an unsolicited state-change is sent [Robustness
Variable] total times so the membership change survives the loss of up to
RV-1 packets, mirroring the shipped IGMPv3 retransmit train.

New on the ICMPv6 TX handler (the IPv6 analogue of the IGMP
'_igmp_state_change__pending' machinery):
- '_MldPendingChange' (the source-bearing §6.1 difference records to
  re-send in MLDv2 mode, the reception-edge direction for the MLDv1 coarse
  Report/Done fallback, and the remaining-repeat count) +
  '_mld_state_change__pending' / '_mld_state_change__handle'.
- '_send_mld_state_change' now, after the immediate emission, seeds a
  pending entry with RV-1 repeats and arms a retransmit ticket (a fresh
  change supersedes any train pending for the same group).
- '_arm' schedules one ticket at random(1, mld.unsolicited_report_interval]
  ms via 'stack.timer'; '_fire' re-emits under '_lock__multicast' in the
  live compat mode (MLDv2 coalesces all pending groups into one Report,
  MLDv1 emits the coarse form per group), decrements each entry, drops the
  exhausted, and re-arms while any remain; '_cancel' drops the ticket +
  clears the map on a compat-mode change (RFC 3810 §8.2.1), wired into the
  RX 'ip_arm_v1_compatibility' MLDv2→MLDv1 transition (mirroring the v4
  '_igmp_cancel_pending_timers' hook).

Three MLD policy sysctls back the train: 'mld.robustness' (RV, default 2),
'mld.unsolicited_report_interval' (default 1000 ms), and 'mld.query_interval'
(default 125 s). The on-touch migration folded the hardcoded
'MLD__ROBUSTNESS_VARIABLE' / 'MLD__QUERY_INTERVAL__MS' module locals out of
'packet_handler__icmp6__rx.py' into the sysctl-backed 'mld__constants',
read via qualified module access so an operator override resolves live —
removing the duplicate definition and completing the MLD constants file's
policy-knob migration.

Tests-first: three retransmit tests added to
'test__icmp6__mld__source_state_change.py' — the RV-1 retransmit carries
the same source-bearing record as the immediate Report, a RV of 2
schedules exactly one retransmit (exhaustion), and an inbound MLDv1 Query
cancels the pending train. The random spacing is patched to a fixed delay
and the FakeTimer advanced to fire the ticket.

Lint clean, 13537 passing, 0 skipped.

Reference: RFC 3810 §6.1 (state-change Report retransmitted RV-1 times).
Reference: RFC 3810 §9.1 (Robustness Variable).
Reference: RFC 3810 §8.2.1 (compat-mode change cancels pending retransmits).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: ccie18643

docs/refactor/host_refinements_backlog.md

@@ -115,6 +115,18 @@ until the user says "push".
   removed. Single immediate emission — the §6.1 robustness retransmit train
   is P4b. Tests: `test__icmp6__mld__source_state_change.py` (7). RFC 3810
   §4.2/§5.2 adherence flip waits for P4b + P5.
+- [x] **IPv6 SSM robustness retransmit train (R4 P4b)** — the RFC 3810 §6.1
+  state-change retransmit machine. `_send_mld_state_change` now schedules
+  RV-1 retransmits (`_MldPendingChange` + `_mld_state_change__pending` +
+  `_arm`/`_fire`/`_cancel_mld_state_change_retransmit(s)`, mirroring the v4
+  IGMP train) at random(1, `mld.unsolicited_report_interval`] via
+  `stack.timer`, re-arming per fire; a compat-mode change (MLDv1 Query)
+  cancels the train. Added three MLD policy sysctls (`mld.robustness` /
+  `mld.unsolicited_report_interval` / `mld.query_interval`), folding the
+  hardcoded `MLD__ROBUSTNESS_VARIABLE` / `MLD__QUERY_INTERVAL__MS` locals
+  in `packet_handler__icmp6__rx.py` into the sysctl-backed `mld__constants`
+  (on-touch migration). Tests: 3 retransmit tests. P5 (RX source-delivery
+  gate) is the last piece before the §4.2/§5.2 adherence flip.
 
 **The canonical SO_RCVBUF guard pattern** (mirror for any new datagram socket):
 
@@ -268,8 +280,20 @@ self._signal_readable()
     `_arm`/`_fire`/`_cancel` machinery). Tests-first:
     `test__icmp6__mld__source_state_change.py` (7) — ALLOW / BLOCK / TO_EX /
     TO_IN records with source lists, mirroring the v4 track.
-  - P4b: the §6.1 robustness retransmit-train state machine
-    (`_mld_state_change__pending` + `_arm`/`_fire`/`_cancel`), mirror v4.
+  - **P4b — SHIPPED:** the §6.1 robustness retransmit-train state machine.
+    Added `_MldPendingChange` + `_mld_state_change__pending` /
+    `_mld_state_change__handle` on the ICMPv6 TX handler and
+    `_arm`/`_fire`/`_cancel_mld_state_change_retransmit(s)` mirroring the v4
+    IGMP machinery: `_send_mld_state_change` now schedules RV-1 retransmits
+    at random(1, `mld.unsolicited_report_interval`] via `stack.timer`, the
+    ticket re-arms per fire, and a compat-mode change (MLDv1 Query) cancels
+    the train (wired into `_mld_arm_v1_compatibility`). Three new MLD
+    policy sysctls (`mld.robustness` / `mld.unsolicited_report_interval` /
+    `mld.query_interval`); the on-touch migration folded the hardcoded
+    `MLD__ROBUSTNESS_VARIABLE` / `MLD__QUERY_INTERVAL__MS` locals out of
+    `packet_handler__icmp6__rx.py` into the sysctl-backed `mld__constants`
+    (qualified module access). Tests: 3 retransmit tests added to
+    `test__icmp6__mld__source_state_change.py`.
   - P5: RX source-delivery filter for IPv6 UDP + RAW (`Ip6MulticastFilter.allows`).
   - Adherence: update `docs/rfc/icmp6/rfc3810__mld2/adherence.md` (§4.2.12 /
     §5.1 / §5.2 source records) + a socket-parity note, in lockstep.
@@ -284,13 +308,13 @@ core (`_Ip6GroupMembership` / `_ip6_multicast_filters` / `_ip6_multicast_refs`
 / `mc6_*` / `_ip6_multicast_filter_for`, all under `_lock__multicast`) AND
 the socket surface (`stack.membership6`, refcounted `IPV6_JOIN_GROUP`, the
 `MCAST_*_SOURCE_*` family with `group_source_req`, per-socket
-`_ip6_source_filters`, close-time release) are done. **P4a SHIPPED** — the
+`_ip6_source_filters`, close-time release) are done. **P4a + P4b SHIPPED** — the
 `_send_mld_state_change` source-bearing state-change records (ALLOW / BLOCK /
-CHANGE_TO_*) now fire on all three edges (join / leave / mid-delta),
-single-emission. **Remaining: P4b** (the §6.1 robustness retransmit-train
-state machine — mirror the v4 `_igmp_state_change__pending` /
-`_arm`/`_fire`/`_cancel`), **P5** (RX `Ip6MulticastFilter.allows` source-
-delivery gate for UDP + RAW). After P4b/P5 flip the RFC 3810 §4.2 / §5.2
+CHANGE_TO_*) fire on all three edges (join / leave / mid-delta) AND
+retransmit RV-1 times per the §6.1 robustness train (`_MldPendingChange` +
+`_arm`/`_fire`/`_cancel`, cancelled on a compat-mode change; three new
+`mld.*` sysctls). **Remaining: P5** (RX `Ip6MulticastFilter.allows` source-
+delivery gate for UDP + RAW). After P5 flip the RFC 3810 §4.2 / §5.2
 adherence rows.
 
 - **Handler (`runtime/packet_handler/__init__.py`) — the deep, no-GIL-locked

packages/pytcp/pytcp/protocols/icmp6/mld__constants.py

@@ -38,11 +38,30 @@
 # MLDv2 regardless of the queriers heard.
 MLD__FORCE_VERSION = 0
 
-# Sysctl registration. The constant above is a policy knob,
+# RFC 3810 §9.1 Robustness Variable (default 2). A host transmits an
+# unsolicited state-change Report RV total times (the initial Report plus
+# RV-1 retransmits) so the membership change survives the loss of up to
+# RV-1 packets. The IPv6 analogue of 'IGMP__ROBUSTNESS_VARIABLE'.
+MLD__ROBUSTNESS_VARIABLE = 2
+
+# RFC 3810 §9.11 Unsolicited Report Interval, in milliseconds (the RFC
+# default is 1 second). Each state-change-Report retransmit is spaced by
+# a value drawn uniformly at random from (0, this interval]. The IPv6
+# analogue of 'IGMP__UNSOLICITED_REPORT_INTERVAL__MS'.
+MLD__UNSOLICITED_REPORT_INTERVAL__MS = 1000
+
+# RFC 3810 §9.2 Query Interval default (125 s), in milliseconds. An MLDv1
+# Query carries no QQIC, so this default is the [Query Interval] term of
+# the §9.12 Older Version Querier Present Timeout the host arms when it
+# hears such a Query. The IPv6 analogue of 'IGMP__QUERY_INTERVAL__MS'.
+MLD__QUERY_INTERVAL__MS = 125_000
+
+# Sysctl registration. Every constant above is a policy knob,
 # operator-tunable at boot via 'stack.init(sysctls={...})' or at
-# runtime via 'pytcp.stack.sysctl["mld.version"] = N'.
+# runtime via 'pytcp.stack.sysctl["mld...."] = N'.
 from pytcp.stack.sysctl import (  # noqa: E402
     is_int_in_range,
+    is_positive_int,
     register,
 )
 
@@ -54,3 +73,27 @@
     validator=is_int_in_range("mld.version", low=0, high=2),
     description="RFC 3810 §8 forced MLD Host Compatibility Mode — 0 = auto fallback, 1/2 = pin MLDv1/MLDv2.",
 )
+register(
+    key="mld.robustness",
+    module_name=__name__,
+    attr="MLD__ROBUSTNESS_VARIABLE",
+    default=MLD__ROBUSTNESS_VARIABLE,
+    validator=is_positive_int("mld.robustness"),
+    description="RFC 3810 §9.1 Robustness Variable — unsolicited state-change Report transmission count.",
+)
+register(
+    key="mld.unsolicited_report_interval",
+    module_name=__name__,
+    attr="MLD__UNSOLICITED_REPORT_INTERVAL__MS",
+    default=MLD__UNSOLICITED_REPORT_INTERVAL__MS,
+    validator=is_positive_int("mld.unsolicited_report_interval"),
+    description="RFC 3810 §9.11 Unsolicited Report Interval — state-change Report retransmit spacing, ms.",
+)
+register(
+    key="mld.query_interval",
+    module_name=__name__,
+    attr="MLD__QUERY_INTERVAL__MS",
+    default=MLD__QUERY_INTERVAL__MS,
+    validator=is_positive_int("mld.query_interval"),
+    description="RFC 3810 §9.2 Query Interval — [Query Interval] term of the MLDv1 querier-present timeout, ms.",
+)

packages/pytcp/pytcp/runtime/packet_handler/packet_handler__icmp6__rx.py

@@ -66,6 +66,7 @@
 from pytcp import stack
 from pytcp.lib.dad_slot_registry import DadSignalResult
 from pytcp.lib.logger import log
+from pytcp.protocols.icmp6 import mld__constants
 from pytcp.protocols.icmp6.icmp6__echo_gate import should_emit_echo_reply
 from pytcp.protocols.icmp.icmp__error_demux import EmbeddedL4, parse_embedded_l4
 from pytcp.protocols.tcp.tcp__icmp_metadata import IcmpCategory, IcmpMetadata
@@ -82,14 +83,6 @@
     from pytcp.runtime.packet_handler import PacketHandler
 
 
-# RFC 3810 §9.1 / §9.2 — MLD Robustness Variable and Query Interval
-# defaults, used to compute the §9.12 Older Version Querier Present
-# Timeout = [Robustness Variable] x [Query Interval] + [Query Response
-# Interval] when an MLDv1 Query arms the compatibility timer.
-MLD__ROBUSTNESS_VARIABLE = 2
-MLD__QUERY_INTERVAL__MS = 125_000
-
-
 def _mld2_mrc_to_mrd_ms(mrc: int) -> int:
     """
     RFC 3810 §5.1.3 — convert a 16-bit Maximum Response
@@ -1233,9 +1226,16 @@ def _mld_arm_v1_compatibility(self, max_response_delay_ms: int, /) -> None:
         arming.
         """
 
-        timeout_ms = MLD__ROBUSTNESS_VARIABLE * MLD__QUERY_INTERVAL__MS + max_response_delay_ms
+        timeout_ms = (
+            mld__constants.MLD__ROBUSTNESS_VARIABLE * mld__constants.MLD__QUERY_INTERVAL__MS + max_response_delay_ms
+        )
         with self._if._lock__multicast:
+            old_mode = self._if._mld_host_compatibility_mode()
             self._if._mld__v1_querier_present_until_ms = stack.timer.now_ms + timeout_ms
+            # RFC 3810 §8.2.1 — a compatibility-mode change (MLDv2→MLDv1)
+            # cancels every pending state-change retransmission.
+            if self._if._mld_host_compatibility_mode() is not old_mode:
+                self._if._icmp6_tx._cancel_mld_state_change_retransmits()
 
     def _mld_query__schedule_response(self, mrd_ms: int, /) -> None:
         """

packages/pytcp/pytcp/runtime/packet_handler/packet_handler__icmp6__tx.py

@@ -32,7 +32,9 @@
 ver 3.0.8
 """
 
+import random
 import struct
+from dataclasses import dataclass
 from typing import TYPE_CHECKING
 
 from net_addr import Buffer, Ip6Address
@@ -71,12 +73,15 @@
     Ip6HbhOptionRouterAlert,
 )
 from net_proto.protocols.ip6_hbh.options.ip6_hbh__options import Ip6HbhOptions
+from pytcp import stack
 from pytcp.lib.ip6_multicast_filter import (
     Ip6MulticastFilter,
     Ip6MulticastFilterMode,
 )
 from pytcp.lib.logger import log
 from pytcp.lib.tx_status import TxStatus
+from pytcp.protocols.icmp6 import mld__constants
+from pytcp.runtime.timer import TimerHandle
 from pytcp.stack import sysctl_iface
 
 if TYPE_CHECKING:
@@ -89,19 +94,40 @@
 MLD__ALL_NODES = Ip6Address("ff02::1")
 
 
+@dataclass(frozen=True, kw_only=True, slots=True)
+class _MldPendingChange:
+    """
+    A pending MLDv2 state-change for one group awaiting robustness
+    retransmission — the source-bearing §6.1 difference records to
+    re-send in MLDv2 mode, the reception-edge direction to re-send in
+    MLDv1 compatibility mode ('True' for a join → MLDv1 Report, 'False'
+    for a leave → MLDv1 Done, 'None' for a source-only change an
+    older-version host cannot express), and the number of
+    retransmissions still owed. The IPv6 analogue of '_IgmpPendingChange'.
+    """
+
+    records: tuple[Icmp6Mld2MulticastAddressRecord, ...]
+    coarse_join: bool | None
+    remaining: int
+
+
 class Icmp6TxHandler:
     """
     The outbound ICMPv6 packet handler for one interface.
     """
 
     _if: PacketHandler
+    _mld_state_change__pending: dict[Ip6Address, _MldPendingChange]
+    _mld_state_change__handle: TimerHandle | None
 
     def __init__(self, *, interface: PacketHandler) -> None:
         """
         Bind the handler to its owning interface.
         """
 
         self._if = interface
+        self._mld_state_change__pending = {}
+        self._mld_state_change__handle = None
 
     def _phtx_icmp6(
         self,
@@ -394,6 +420,19 @@ def _mld_state_change_records(
             )
         ]
 
+    def _emit_mld1_coarse(self, group: Ip6Address, coarse_join: bool | None, /) -> None:
+        """
+        Emit the coarse MLDv1 form of a state-change for 'group' while in
+        MLDv1 Host Compatibility Mode: a Report on a join edge, a Done on a
+        leave edge, nothing for a source-only change (MLDv1 has no source
+        concept). 'coarse_join' is True (join), False (leave) or None.
+        """
+
+        if coarse_join is True:
+            self._send_icmp6_mld1_report(group)
+        elif coarse_join is False:
+            self._send_icmp6_mld1_done(group)
+
     def _send_mld_state_change(
         self,
         group: Ip6Address,
@@ -406,27 +445,125 @@ def _send_mld_state_change(
         Emit an unsolicited state-change Report for 'group' describing the
         transition from filter 'old' to filter 'new' (RFC 3810 §6.1) in
         the form dictated by the interface's Host Compatibility Mode
-        (§8.3). In MLDv2 mode the Report carries the source-bearing §6.1
-        difference records; in MLDv1 mode it degrades to the coarse
-        Report / Done (MLDv1 has no source concept), keyed only off the
-        reception edge. The all-nodes group ff02::1 is never reported
-        (RFC 3810 §6). The IPv6 analogue of '_send_igmp_state_change'.
+        (§8.3), and schedule its robustness retransmissions. In MLDv2 mode
+        the Report carries the source-bearing §6.1 difference records; in
+        MLDv1 mode it degrades to the coarse Report / Done keyed only off
+        the reception edge (MLDv1 has no source concept). The all-nodes
+        group ff02::1 is never reported (RFC 3810 §6).
+
+        A new change supersedes any retransmit train still pending for the
+        same group (overwrite + re-seed). A change that produces no record
+        (an idempotent re-add) schedules no retransmit. The IPv6 analogue
+        of '_send_igmp_state_change'.
         """
 
         if group == MLD__ALL_NODES:
             return
 
+        records = self._mld_state_change_records(group, old, new)
         # The coarse MLDv1 form keys only off the reception edge — a
-        # source-only change within a still-joined membership is invisible
-        # to an older-version querier.
+        # source-only change within a still-joined membership (coarse_join
+        # None) is invisible to an older-version querier.
+        coarse_join = (
+            True
+            if new.has_reception and not old.has_reception
+            else False if old.has_reception and not new.has_reception else None
+        )
+
         if self._if._mld_host_compatibility_mode() is MldVersion.V1:
-            if new.has_reception and not old.has_reception:
-                self._send_icmp6_mld1_report(group)
-            elif old.has_reception and not new.has_reception:
-                self._send_icmp6_mld1_done(group)
+            self._emit_mld1_coarse(group, coarse_join)
+        else:
+            self._emit_mld2_report(records)
+
+        # RFC 3810 §9.1 — read the Robustness Variable live so an operator
+        # override resolves per change.
+        repeats = mld__constants.MLD__ROBUSTNESS_VARIABLE - 1
+        if not records or repeats <= 0:
+            self._mld_state_change__pending.pop(group, None)
+            return
+
+        self._mld_state_change__pending[group] = _MldPendingChange(
+            records=tuple(records),
+            coarse_join=coarse_join,
+            remaining=repeats,
+        )
+        self._arm_mld_state_change_retransmit()
+
+    def _arm_mld_state_change_retransmit(self) -> None:
+        """
+        Ensure a single retransmit ticket is scheduled for the pending
+        state-change records. RFC 3810 §6.1 spaces the robustness
+        retransmissions at intervals drawn uniformly at random from (0,
+        'mld.unsolicited_report_interval' ms]; the ticket re-arms itself
+        from each fire rather than scheduling the whole train up front, so
+        a change arriving mid-train is picked up by the next fire. Reading
+        the interval knob via qualified module access so an operator
+        override resolves on each re-arm. The IPv6 analogue of
+        '_arm_state_change_retransmit'.
+        """
+
+        if self._mld_state_change__handle is not None:
             return
 
-        self._emit_mld2_report(self._mld_state_change_records(group, old, new))
+        delay_ms = random.randint(1, mld__constants.MLD__UNSOLICITED_REPORT_INTERVAL__MS)
+        self._mld_state_change__handle = stack.timer.call_later(delay_ms, self._fire_mld_state_change_retransmit)
+
+    def _fire_mld_state_change_retransmit(self) -> None:
+        """
+        Emit one robustness retransmission of the currently-pending
+        state-change records in the current Host Compatibility Mode's form
+        — MLDv2 coalesces them into a single Report, MLDv1 emits one
+        coarse Report / Done per group — then decrement each entry's
+        remaining-repeat count, drop the exhausted ones, and re-arm the
+        ticket while any repeats remain (RFC 3810 §6.1 / §8.3). The IPv6
+        analogue of '_fire_state_change_retransmit'.
+        """
+
+        # Runs on the Timer thread. The pending per-group change map and
+        # the Host Compatibility Mode deadlines are mutated / written by
+        # the RX and application threads under the interface multicast
+        # lock, so this fire takes the same lock (the RLock is reentrant,
+        # so the nested compat-mode / emit reads are fine).
+        with self._if._lock__multicast:
+            self._mld_state_change__handle = None
+
+            groups = list(self._mld_state_change__pending)
+
+            if self._if._mld_host_compatibility_mode() is MldVersion.V1:
+                for group in groups:
+                    self._emit_mld1_coarse(group, self._mld_state_change__pending[group].coarse_join)
+            else:
+                records: list[Icmp6Mld2MulticastAddressRecord] = []
+                for group in groups:
+                    records.extend(self._mld_state_change__pending[group].records)
+                self._emit_mld2_report(records)
+
+            for group in groups:
+                pending = self._mld_state_change__pending[group]
+                if pending.remaining <= 1:
+                    del self._mld_state_change__pending[group]
+                else:
+                    self._mld_state_change__pending[group] = _MldPendingChange(
+                        records=pending.records,
+                        coarse_join=pending.coarse_join,
+                        remaining=pending.remaining - 1,
+                    )
+
+            if self._mld_state_change__pending:
+                self._arm_mld_state_change_retransmit()
+
+    def _cancel_mld_state_change_retransmits(self) -> None:
+        """
+        Cancel the in-flight state-change retransmit ticket and drop every
+        pending per-group change record (RFC 3810 §8.2.1 — a
+        compatibility-mode change cancels all pending retransmissions). The
+        IPv6 analogue of '_cancel_state_change_retransmits'.
+        """
+
+        if self._mld_state_change__handle is not None:
+            stack.timer.cancel(self._mld_state_change__handle)
+            self._mld_state_change__handle = None
+        self._mld_state_change__pending.clear()
 
     def __send_icmp6_mld_via_hbh_ra(self, icmp6_packet_tx: Icmp6Assembler, /, *, ip6__dst: Ip6Address) -> None:
         """