diff --git a/CHANGELOG.md b/CHANGELOG.md index bd9b494a..876b3b0b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,41 @@ This file holds unreleased changes and the current release. Older releases are archived by series under [docs/changelog/](docs/changelog/); see the [archive index](docs/changelog/README.md). +## [Unreleased] + +### Fixed + +- **Internet-only providers no longer flap a relay's rate limiter into a + disconnect loop under a backlog of unreachable direct messages.** Three + compounding defects drove unbounded resends to gone peers. (1) A resend that + had to register a fresh ACK (for example after an ACK-timeout re-queue) + restarted the backoff ladder at `retry_count` 0, pinning `delay_for_retry` at + its 1s floor forever for a never-ACKing recipient; + `AckManager::set_retry_count` now carries the retry-queue entry's accumulated + count onto the fresh ACK so backoff continues instead of resetting. (2) An + unconfirmed peer that vanished was re-probed every 5s indefinitely; the + confirmation probe now escalates on the same 15s to 600s ladder as the welcome + lifecycle and resets on a reachability edge. (3) The Python relay bridge + dropped the relay's recipient-keyed `DeliveryError` verdict; it now correlates + in-flight sends per recipient (a port of the iOS/Android + `RecipientInFlightTracker`), fails the affected message ids fast, and feeds + `internet_peer_presence(online=false)`. Default behavior is otherwise + unchanged; the always-on fixes only reduce redundant relay traffic. + +### Added + +- **`RetryConfig.edge_driven_unreachable_dm`** (default `false`): an opt-in that, + for deployments whose peers always interact or advertise presence on return + (for example a machine-to-machine capability exchange), stops timed-probing a + durably-unreachable direct message after a bounded number of probes and + re-drives it only on a reachability edge, additionally bounding the core resend + rate to gone peers. Default `false` preserves the documented perpetual-probe + contract (a parked message "never goes fully quiet") so every existing native + and third-party integration is unaffected. Exposed across the UDL, all + generated bindings (Swift, Kotlin, Python), and the React Native TypeScript and + native layers. See + [docs/configuration.md](docs/configuration.md#reliability-configuration). + ## [0.24.1] — 2026-08-26 > **Every fix here is the same defect found four ways: the iOS React Native diff --git a/bindings/python/offline_protocol_sdk/internet_manager.py b/bindings/python/offline_protocol_sdk/internet_manager.py index 445bea5a..1a596aaf 100644 --- a/bindings/python/offline_protocol_sdk/internet_manager.py +++ b/bindings/python/offline_protocol_sdk/internet_manager.py @@ -11,6 +11,8 @@ import base64 import json import logging +import time +from collections import deque from typing import Any, Coroutine import websockets @@ -32,6 +34,103 @@ _MAX_CONSECUTIVE_FAILURES = 2 _MAX_CONCURRENT_SENDS = 50 +# In-flight tracker tuning (mirrors iOS/Android RecipientInFlightTracker). +_RIFT_TTL_MS = 60_000 +_RIFT_MAX_PER_RECIPIENT = 32 + + +class _RecipientInFlightTracker: + """Python port of the iOS/Android ``RecipientInFlightTracker``. + + Tracks wire-level in-flight message ids per recipient so the relay's + recipient-keyed failure signal — ``DeliveryError``, which on older relays + carries no ``message_id`` — can be correlated back to the SDK message ids + still awaiting an outcome. On a relay new enough to echo the outbox id we + resolve delivered frames precisely on ``MessageSent`` and fail the exact id + on ``DeliveryError``; on an older relay we fall back to failing every live + in-flight id for the recipient. "Everything in flight to an offline peer + failed" is safe by construction. Runs on the single asyncio loop, so no lock + is needed (unlike the mobile bridges). + """ + + def __init__(self, ttl_ms: int = _RIFT_TTL_MS, max_per_recipient: int = _RIFT_MAX_PER_RECIPIENT) -> None: + self._ttl_ms = ttl_ms + self._max = max_per_recipient + self._by_recipient: dict[str, deque[tuple[str, int]]] = {} + + def record_sent(self, recipient: str, message_id: str, now_ms: int) -> None: + if not recipient or not message_id: + return + q = self._by_recipient.setdefault(recipient, deque()) + q.append((message_id, now_ms)) + while len(q) > self._max: + q.popleft() + + def resolve_on_relay_accepted(self, recipient: str, message_id: str | None, now_ms: int) -> None: + """Relay ``MessageSent``: it accepted/forwarded a frame, so that frame + must not be swept into a later recipient-keyed ``DeliveryError`` (which + would false-fail a delivered message). Remove the exact id when the relay + echoed ours; otherwise (older relay / relay-minted id) drop the oldest as + a best-effort guess, bounded by the TTL and by the DeliveryError sweep.""" + if not recipient: + return + q = self._by_recipient.get(recipient) + if q is None: + return + while q and now_ms - q[0][1] > self._ttl_ms: + q.popleft() + exact = False + if message_id: + q2 = deque((m, t) for (m, t) in q if m != message_id) + exact = len(q2) != len(q) + q = q2 + self._by_recipient[recipient] = q + if not exact and q: + q.popleft() + if not q: + self._by_recipient.pop(recipient, None) + + def unrecord(self, recipient: str, message_id: str) -> None: + """Undo a ``record_sent`` when the write never reached the wire. + + The entry is recorded BEFORE ``await ws.send`` so a fast relay + ``DeliveryError`` interleaved on the recv task finds it; when the send + instead raises, the frame did not go out and its optimistic entry must + be taken back, or a later recipient-keyed ``DeliveryError`` would + false-fail a message that was never in flight. Mirrors the iOS/Android + failure-completion ``unrecord``.""" + if not recipient or not message_id: + return + q = self._by_recipient.get(recipient) + if q is None: + return + # Remove the newest matching id (the one this call just recorded), + # leaving any older same-id retry entry intact. + for i in range(len(q) - 1, -1, -1): + if q[i][0] == message_id: + del q[i] + break + if not q: + self._by_recipient.pop(recipient, None) + + def drain_recipient(self, recipient: str, now_ms: int) -> list[str]: + """Remove and return every live (non-expired) in-flight id for a peer.""" + q = self._by_recipient.pop(recipient, None) + if q is None: + return [] + return [m for (m, t) in q if now_ms - t <= self._ttl_ms] + + def prune(self, now_ms: int) -> None: + for r in list(self._by_recipient.keys()): + q = self._by_recipient[r] + while q and now_ms - q[0][1] > self._ttl_ms: + q.popleft() + if not q: + self._by_recipient.pop(r, None) + + def clear(self) -> None: + self._by_recipient.clear() + class InternetManager(TransportManager): """WebSocket-based internet transport. @@ -110,6 +209,9 @@ def __init__( # weak task table cannot GC them mid-execution. self._process_tasks: set[asyncio.Task[None]] = set() self._send_semaphore: asyncio.Semaphore = asyncio.Semaphore(_MAX_CONCURRENT_SENDS) + # Correlates the relay's recipient-keyed DeliveryError back to in-flight + # sends (parity with the iOS/Android bridges). + self._inflight = _RecipientInFlightTracker() self._reconnect_handle: asyncio.TimerHandle | None = None # Re-entrancy guard for `_handle_connection_closed`. Set synchronously @@ -478,6 +580,11 @@ async def _handle_connection_closed(self, error: Exception | None) -> None: self._connected = False self._authenticated = False + # The socket died: forget in-flight correlations. Anything still + # unresolved is owned by the transport/core retry machinery now, and + # a fresh connection re-records from scratch. + self._inflight.clear() + # Cancel recv/poll/ping tasks. Skip whichever (if any) is the # task currently running this coroutine — any task in the cancel # list can itself be the caller (recv-loop on ConnectionClosed, @@ -684,13 +791,65 @@ def _process_received(self, data: bytes) -> None: elif msg_type == "ConnectionRejected": self._handle_connection_rejected(msg) + elif msg_type == "MessageSent": + # The relay accepted/forwarded this frame (or push-poked an offline + # recipient) — either way it is no longer in flight and must not be + # swept into a later recipient-keyed DeliveryError, which would + # false-fail a delivered message. Resolve it out of the in-flight + # tracker (by exact id when the relay echoed ours, else best-effort + # oldest). Not a delivery guarantee, so we do not touch presence. + recipient = msg.get("recipient", "") + message_id = msg.get("message_id") + if recipient: + self._inflight.resolve_on_relay_accepted( + recipient, + message_id if message_id else None, + self._now_ms(), + ) + elif msg_type == "DeliveryError": - recipient = msg.get("recipient", "unknown") + recipient = msg.get("recipient", "") + message_id = msg.get("message_id") reason = msg.get("reason", "unknown") self._emit_diagnostic("warning", "Delivery failed", { "recipient": recipient, "reason": reason, }) + # The relay's authoritative "recipient offline" signal. Feed it back + # into the core tagged with the "recipient_unreachable" prefix the + # engine classifies on (see SEND_FAIL_REASON_RECIPIENT_UNREACHABLE) + # so the message is parked on the escalating reachability probe + # instead of re-sent ~once/second forever (which would exceed the + # relay's per-connection rate limit and flap the connection). + # + # Correlate by RECIPIENT, matching the iOS/Android bridges: the relay + # is recipient-keyed and older relays send no message_id at all, so + # we fail every live in-flight id for this recipient (delivered ones + # were already resolved out on their MessageSent). On a relay new + # enough to echo the outbox id we additionally fail that exact id in + # case it was never recorded. Then feed presence-offline so the core + # parks welcomes and starts watching for the peer's return. + now = self._now_ms() + failed_ids = self._inflight.drain_recipient(recipient, now) if recipient else [] + if message_id and message_id not in failed_ids: + failed_ids.append(message_id) + for mid in failed_ids: + try: + self._protocol.internet_send_failed_with_reason( + message_id=mid, + reason=f"recipient_unreachable: {reason}", + ) + except Exception as exc: + logger.debug( + "internet_send_failed_with_reason failed: %s", exc + ) + if recipient: + try: + self._protocol.internet_peer_presence( + peer_id=recipient, online=False, last_seen_ms=None + ) + except Exception as exc: + logger.debug("internet_peer_presence(offline) failed: %s", exc) elif msg_type in ("GroupCreated", "GroupInvitation", "GroupMessageReceived"): self._handle_group_message(msg_type, msg) @@ -859,6 +1018,9 @@ def _poll_and_send_messages(self) -> None: prevent unbounded task creation when the protocol core has a large outbox. Remaining messages will be picked up on the next poll tick. """ + # Drop in-flight tracker entries older than the TTL each tick. + self._inflight.prune(self._now_ms()) + # Limit to available concurrency slots to avoid creating thousands # of tasks that all block on the semaphore. available_slots = max(0, _MAX_CONCURRENT_SENDS - len(self._send_tasks)) @@ -886,6 +1048,10 @@ def _poll_and_send_messages(self) -> None: task.add_done_callback(self._send_tasks.discard) drained += 1 + @staticmethod + def _now_ms() -> int: + return int(time.monotonic() * 1000) + def _notify_send_failed(self, message_id: str, reason: str) -> None: """Best-effort notification to the protocol that a send failed.""" try: @@ -914,6 +1080,7 @@ async def _send_message( self._notify_send_failed(message_id, "Disconnected while waiting") return + recorded = False try: try: content = data.decode("utf-8") @@ -932,7 +1099,26 @@ async def _send_message( "content": content, "message_id": message_id, }) + # Record the wire send BEFORE awaiting it: `await ws.send` can + # suspend (websockets drains under backpressure — the exact + # high-load path this correlation exists for), and a relay + # MessageSent/DeliveryError for this frame can be processed on + # the recv task before we resume. The entry must already be + # present so MessageSent resolves the exact id (rather than + # best-effort popping a different still-stuck one) and a + # DeliveryError finds it. If the send raises, the except path + # unrecords it. Mirrors the iOS/Android record-before-write. + if recipient and message_id: + self._inflight.record_sent(recipient, message_id, self._now_ms()) + recorded = True await ws.send(payload) + # The frame is on the wire now. From here `unrecord` must NOT + # fire: the entry is genuinely in flight, and a later relay + # DeliveryError needs it to fast-fail. Clear the flag before any + # subsequent line (counters, the internet_confirm_sent FFI call) + # can raise, so only a failure of `await ws.send` itself — where + # the frame never left — takes the entry back. + recorded = False self._bytes_sent += len(payload) self._messages_sent += 1 self._consecutive_send_failures = 0 @@ -940,6 +1126,12 @@ async def _send_message( self._protocol.internet_confirm_sent(message_id=message_id) except Exception as exc: + # The frame never reached the wire; take back the optimistic + # in-flight entry so a later recipient-keyed DeliveryError does + # not false-fail a message that was never sent. (Only reachable + # for a pre-wire failure; a post-wire raise leaves it in flight.) + if recorded: + self._inflight.unrecord(recipient, message_id) self._consecutive_send_failures += 1 self._emit_diagnostic("error", f"Send failed: {exc}", { "message_id": message_id, diff --git a/bindings/python/offline_protocol_sdk/offline_protocol.py b/bindings/python/offline_protocol_sdk/offline_protocol.py index 7f5449f0..650c2447 100644 --- a/bindings/python/offline_protocol_sdk/offline_protocol.py +++ b/bindings/python/offline_protocol_sdk/offline_protocol.py @@ -6332,7 +6332,7 @@ def read(cls, buf): @dataclass class ProtocolConfig: - def __init__(self, *, app_id:str, profile:str, ble_enabled:bool, wifi_direct_enabled:bool, internet_enabled:bool, reticulum_enabled:bool, nostr_enabled:bool, prefer_online:bool, initial_ttl:int, encryption_enabled:bool, auto_key_exchange:bool, store_pending:bool, require_encryption:bool = True, max_pending_per_peer:int, max_pending_global:int, pending_ttl_ms:int, overflow_policy:OverflowPolicy, max_group_members:int = 256, group_relay_enabled:bool = True, group_relay_broadcast_enabled:bool = True, group_enforce_admin_commits:bool = False, require_transport_identity:bool = False, binary_wire_enabled:bool = True, nostr_sealing_enabled:bool = True, nostr_cold_contact_enabled:bool = True, nostr_username_discovery_enabled:bool = False, compact_envelope_enabled:bool = True, rich_payload_enabled:bool = True, crypto_recovery_enabled:bool = True, mesh_relay:typing.Optional[MeshRelayConfig] = _DEFAULT, data_enabled:bool = True, control_freshness_enforced:bool = True): + def __init__(self, *, app_id:str, profile:str, ble_enabled:bool, wifi_direct_enabled:bool, internet_enabled:bool, reticulum_enabled:bool, nostr_enabled:bool, prefer_online:bool, initial_ttl:int, encryption_enabled:bool, auto_key_exchange:bool, store_pending:bool, require_encryption:bool = True, max_pending_per_peer:int, max_pending_global:int, pending_ttl_ms:int, overflow_policy:OverflowPolicy, edge_driven_unreachable_dm:bool = False, max_group_members:int = 256, group_relay_enabled:bool = True, group_relay_broadcast_enabled:bool = True, group_enforce_admin_commits:bool = False, require_transport_identity:bool = False, binary_wire_enabled:bool = True, nostr_sealing_enabled:bool = True, nostr_cold_contact_enabled:bool = True, nostr_username_discovery_enabled:bool = False, compact_envelope_enabled:bool = True, rich_payload_enabled:bool = True, crypto_recovery_enabled:bool = True, mesh_relay:typing.Optional[MeshRelayConfig] = _DEFAULT, data_enabled:bool = True, control_freshness_enforced:bool = True): self.app_id = app_id self.profile = profile self.ble_enabled = ble_enabled @@ -6350,6 +6350,7 @@ def __init__(self, *, app_id:str, profile:str, ble_enabled:bool, wifi_direct_ena self.max_pending_global = max_pending_global self.pending_ttl_ms = pending_ttl_ms self.overflow_policy = overflow_policy + self.edge_driven_unreachable_dm = edge_driven_unreachable_dm self.max_group_members = max_group_members self.group_relay_enabled = group_relay_enabled self.group_relay_broadcast_enabled = group_relay_broadcast_enabled @@ -6373,7 +6374,7 @@ def __init__(self, *, app_id:str, profile:str, ble_enabled:bool, wifi_direct_ena def __str__(self): - return "ProtocolConfig(app_id={}, profile={}, ble_enabled={}, wifi_direct_enabled={}, internet_enabled={}, reticulum_enabled={}, nostr_enabled={}, prefer_online={}, initial_ttl={}, encryption_enabled={}, auto_key_exchange={}, store_pending={}, require_encryption={}, max_pending_per_peer={}, max_pending_global={}, pending_ttl_ms={}, overflow_policy={}, max_group_members={}, group_relay_enabled={}, group_relay_broadcast_enabled={}, group_enforce_admin_commits={}, require_transport_identity={}, binary_wire_enabled={}, nostr_sealing_enabled={}, nostr_cold_contact_enabled={}, nostr_username_discovery_enabled={}, compact_envelope_enabled={}, rich_payload_enabled={}, crypto_recovery_enabled={}, mesh_relay={}, data_enabled={}, control_freshness_enforced={})".format(self.app_id, self.profile, self.ble_enabled, self.wifi_direct_enabled, self.internet_enabled, self.reticulum_enabled, self.nostr_enabled, self.prefer_online, self.initial_ttl, self.encryption_enabled, self.auto_key_exchange, self.store_pending, self.require_encryption, self.max_pending_per_peer, self.max_pending_global, self.pending_ttl_ms, self.overflow_policy, self.max_group_members, self.group_relay_enabled, self.group_relay_broadcast_enabled, self.group_enforce_admin_commits, self.require_transport_identity, self.binary_wire_enabled, self.nostr_sealing_enabled, self.nostr_cold_contact_enabled, self.nostr_username_discovery_enabled, self.compact_envelope_enabled, self.rich_payload_enabled, self.crypto_recovery_enabled, self.mesh_relay, self.data_enabled, self.control_freshness_enforced) + return "ProtocolConfig(app_id={}, profile={}, ble_enabled={}, wifi_direct_enabled={}, internet_enabled={}, reticulum_enabled={}, nostr_enabled={}, prefer_online={}, initial_ttl={}, encryption_enabled={}, auto_key_exchange={}, store_pending={}, require_encryption={}, max_pending_per_peer={}, max_pending_global={}, pending_ttl_ms={}, overflow_policy={}, edge_driven_unreachable_dm={}, max_group_members={}, group_relay_enabled={}, group_relay_broadcast_enabled={}, group_enforce_admin_commits={}, require_transport_identity={}, binary_wire_enabled={}, nostr_sealing_enabled={}, nostr_cold_contact_enabled={}, nostr_username_discovery_enabled={}, compact_envelope_enabled={}, rich_payload_enabled={}, crypto_recovery_enabled={}, mesh_relay={}, data_enabled={}, control_freshness_enforced={})".format(self.app_id, self.profile, self.ble_enabled, self.wifi_direct_enabled, self.internet_enabled, self.reticulum_enabled, self.nostr_enabled, self.prefer_online, self.initial_ttl, self.encryption_enabled, self.auto_key_exchange, self.store_pending, self.require_encryption, self.max_pending_per_peer, self.max_pending_global, self.pending_ttl_ms, self.overflow_policy, self.edge_driven_unreachable_dm, self.max_group_members, self.group_relay_enabled, self.group_relay_broadcast_enabled, self.group_enforce_admin_commits, self.require_transport_identity, self.binary_wire_enabled, self.nostr_sealing_enabled, self.nostr_cold_contact_enabled, self.nostr_username_discovery_enabled, self.compact_envelope_enabled, self.rich_payload_enabled, self.crypto_recovery_enabled, self.mesh_relay, self.data_enabled, self.control_freshness_enforced) def __eq__(self, other): if self.app_id != other.app_id: return False @@ -6409,6 +6410,8 @@ def __eq__(self, other): return False if self.overflow_policy != other.overflow_policy: return False + if self.edge_driven_unreachable_dm != other.edge_driven_unreachable_dm: + return False if self.max_group_members != other.max_group_members: return False if self.group_relay_enabled != other.group_relay_enabled: @@ -6462,6 +6465,7 @@ def read(buf): max_pending_global=_UniffiFfiConverterUInt64.read(buf), pending_ttl_ms=_UniffiFfiConverterUInt64.read(buf), overflow_policy=_UniffiFfiConverterTypeOverflowPolicy.read(buf), + edge_driven_unreachable_dm=_UniffiFfiConverterBoolean.read(buf), max_group_members=_UniffiFfiConverterUInt32.read(buf), group_relay_enabled=_UniffiFfiConverterBoolean.read(buf), group_relay_broadcast_enabled=_UniffiFfiConverterBoolean.read(buf), @@ -6498,6 +6502,7 @@ def check_lower(value): _UniffiFfiConverterUInt64.check_lower(value.max_pending_global) _UniffiFfiConverterUInt64.check_lower(value.pending_ttl_ms) _UniffiFfiConverterTypeOverflowPolicy.check_lower(value.overflow_policy) + _UniffiFfiConverterBoolean.check_lower(value.edge_driven_unreachable_dm) _UniffiFfiConverterUInt32.check_lower(value.max_group_members) _UniffiFfiConverterBoolean.check_lower(value.group_relay_enabled) _UniffiFfiConverterBoolean.check_lower(value.group_relay_broadcast_enabled) @@ -6533,6 +6538,7 @@ def write(value, buf): _UniffiFfiConverterUInt64.write(value.max_pending_global, buf) _UniffiFfiConverterUInt64.write(value.pending_ttl_ms, buf) _UniffiFfiConverterTypeOverflowPolicy.write(value.overflow_policy, buf) + _UniffiFfiConverterBoolean.write(value.edge_driven_unreachable_dm, buf) _UniffiFfiConverterUInt32.write(value.max_group_members, buf) _UniffiFfiConverterBoolean.write(value.group_relay_enabled, buf) _UniffiFfiConverterBoolean.write(value.group_relay_broadcast_enabled, buf) @@ -6689,19 +6695,20 @@ def write(value, buf): @dataclass class RetryConfig: - def __init__(self, *, max_retries:int, initial_delay_ms:int, max_delay_ms:int, backoff_multiplier:float, outbox_max_lifetime_ms:int, pending_message_max_lifetime_ms:int): + def __init__(self, *, max_retries:int, initial_delay_ms:int, max_delay_ms:int, backoff_multiplier:float, outbox_max_lifetime_ms:int, pending_message_max_lifetime_ms:int, edge_driven_unreachable_dm:bool = False): self.max_retries = max_retries self.initial_delay_ms = initial_delay_ms self.max_delay_ms = max_delay_ms self.backoff_multiplier = backoff_multiplier self.outbox_max_lifetime_ms = outbox_max_lifetime_ms self.pending_message_max_lifetime_ms = pending_message_max_lifetime_ms + self.edge_driven_unreachable_dm = edge_driven_unreachable_dm def __str__(self): - return "RetryConfig(max_retries={}, initial_delay_ms={}, max_delay_ms={}, backoff_multiplier={}, outbox_max_lifetime_ms={}, pending_message_max_lifetime_ms={})".format(self.max_retries, self.initial_delay_ms, self.max_delay_ms, self.backoff_multiplier, self.outbox_max_lifetime_ms, self.pending_message_max_lifetime_ms) + return "RetryConfig(max_retries={}, initial_delay_ms={}, max_delay_ms={}, backoff_multiplier={}, outbox_max_lifetime_ms={}, pending_message_max_lifetime_ms={}, edge_driven_unreachable_dm={})".format(self.max_retries, self.initial_delay_ms, self.max_delay_ms, self.backoff_multiplier, self.outbox_max_lifetime_ms, self.pending_message_max_lifetime_ms, self.edge_driven_unreachable_dm) def __eq__(self, other): if self.max_retries != other.max_retries: return False @@ -6715,6 +6722,8 @@ def __eq__(self, other): return False if self.pending_message_max_lifetime_ms != other.pending_message_max_lifetime_ms: return False + if self.edge_driven_unreachable_dm != other.edge_driven_unreachable_dm: + return False return True class _UniffiFfiConverterTypeRetryConfig(_UniffiConverterRustBuffer): @@ -6727,6 +6736,7 @@ def read(buf): backoff_multiplier=_UniffiFfiConverterFloat32.read(buf), outbox_max_lifetime_ms=_UniffiFfiConverterUInt64.read(buf), pending_message_max_lifetime_ms=_UniffiFfiConverterUInt64.read(buf), + edge_driven_unreachable_dm=_UniffiFfiConverterBoolean.read(buf), ) @staticmethod @@ -6737,6 +6747,7 @@ def check_lower(value): _UniffiFfiConverterFloat32.check_lower(value.backoff_multiplier) _UniffiFfiConverterUInt64.check_lower(value.outbox_max_lifetime_ms) _UniffiFfiConverterUInt64.check_lower(value.pending_message_max_lifetime_ms) + _UniffiFfiConverterBoolean.check_lower(value.edge_driven_unreachable_dm) @staticmethod def write(value, buf): @@ -6746,6 +6757,7 @@ def write(value, buf): _UniffiFfiConverterFloat32.write(value.backoff_multiplier, buf) _UniffiFfiConverterUInt64.write(value.outbox_max_lifetime_ms, buf) _UniffiFfiConverterUInt64.write(value.pending_message_max_lifetime_ms, buf) + _UniffiFfiConverterBoolean.write(value.edge_driven_unreachable_dm, buf) @dataclass class ReliabilityConfig: diff --git a/bindings/python/tests/test_internet_manager.py b/bindings/python/tests/test_internet_manager.py index b8f21a8e..3b035203 100644 --- a/bindings/python/tests/test_internet_manager.py +++ b/bindings/python/tests/test_internet_manager.py @@ -570,6 +570,83 @@ async def test_send_message_frame_shape( message_id="msg-1" ) + @pytest.mark.asyncio + async def test_records_in_flight_before_the_wire_write( + self, mock_protocol: MagicMock + ) -> None: + """The in-flight entry must exist BEFORE `await ws.send` returns. + + `await ws.send` can suspend under backpressure and a relay + MessageSent/DeliveryError for the frame can be processed on the recv + task before it resumes; the entry has to already be present so the + exact id resolves rather than best-effort popping a different still- + stuck one. Mirrors the iOS/Android record-before-write ordering. + """ + mgr = InternetManager(mock_protocol, "dev-1", server_url="ws://x.com") + mgr._connected = True + seen_at_write: list[str] = [] + + async def capture(_payload: str) -> None: + # Snapshot what the tracker holds at the instant of the write. + seen_at_write.extend(mgr._inflight.drain_recipient("peer-1", mgr._now_ms())) + # Put it back so the rest of the send path is unaffected. + mgr._inflight.record_sent("peer-1", "msg-1", mgr._now_ms()) + + ws = MagicMock() + ws.send = AsyncMock(side_effect=capture) + mgr._ws = ws + + await mgr._send_message("msg-1", "peer-1", b"hello") + + assert seen_at_write == ["msg-1"], ( + "the send must be recorded before `await ws.send`, not after" + ) + + @pytest.mark.asyncio + async def test_unrecords_in_flight_when_the_write_fails( + self, mock_protocol: MagicMock + ) -> None: + """If `await ws.send` raises, the frame never went out, so its + optimistic in-flight entry must be taken back — otherwise a later + recipient-keyed DeliveryError would false-fail a message never sent.""" + mgr = InternetManager(mock_protocol, "dev-1", server_url="ws://x.com") + mgr._connected = True + ws = MagicMock() + ws.send = AsyncMock(side_effect=RuntimeError("socket broke")) + mgr._ws = ws + + await mgr._send_message("msg-1", "peer-1", b"hello") + + # No residual in-flight entry for the recipient. + assert mgr._inflight.drain_recipient("peer-1", mgr._now_ms()) == [] + # And the failure was reported for the message. + mock_protocol.internet_send_failed_with_reason.assert_called() + assert ( + mock_protocol.internet_send_failed_with_reason.call_args.kwargs["message_id"] + == "msg-1" + ) + + @pytest.mark.asyncio + async def test_post_wire_failure_keeps_the_in_flight_entry( + self, mock_protocol: MagicMock + ) -> None: + """A failure AFTER `await ws.send` succeeded (e.g. the internet_confirm_sent + FFI call raising) must NOT unrecord the entry: the frame is genuinely on + the wire and a later relay DeliveryError needs it to fast-fail. unrecord + fires only when the wire write itself never landed.""" + mgr = InternetManager(mock_protocol, "dev-1", server_url="ws://x.com") + mgr._connected = True + ws = MagicMock() + ws.send = AsyncMock() # the write SUCCEEDS + mgr._ws = ws + # ...but the post-write FFI confirm raises. + mock_protocol.internet_confirm_sent.side_effect = RuntimeError("ffi boom") + + await mgr._send_message("msg-1", "peer-1", b"hello") + + # The entry survives so a later DeliveryError can still fast-fail it. + assert mgr._inflight.drain_recipient("peer-1", mgr._now_ms()) == ["msg-1"] + class TestInternetManagerSendMessageTOCTOU: @pytest.mark.asyncio @@ -876,3 +953,142 @@ def test_the_refusal_frame_is_address_error( declaring_protocol.internet_address_declaration_refused.assert_called_once_with( reason="address_taken" ) + + +# --------------------------------------------------------------------------- +# RecipientInFlightTracker + recipient-keyed DeliveryError correlation. +# Ports the iOS/Android bridge behavior; these pin it per the repo's C9 policy +# ("cargo test proves nothing about the bridges — each binding has its own +# tests"). The highest-risk new logic is the tracker: a subtle error would +# false-fail a delivered message, so the MessageSent-resolve path is tested +# explicitly. +# --------------------------------------------------------------------------- +from offline_protocol_sdk.internet_manager import _RecipientInFlightTracker + + +class TestRecipientInFlightTracker: + def test_record_cap_and_fifo(self) -> None: + t = _RecipientInFlightTracker(ttl_ms=1000, max_per_recipient=3) + for m in ("m1", "m2", "m3", "m4"): + t.record_sent("r1", m, 0) + # oldest dropped at the cap; FIFO order preserved + assert t.drain_recipient("r1", 0) == ["m2", "m3", "m4"] + + def test_resolve_exact_prevents_false_fail(self) -> None: + # A delivered frame (relay echoed our id on MessageSent) must be + # resolved OUT so a later recipient-keyed DeliveryError cannot fail it. + t = _RecipientInFlightTracker(ttl_ms=10_000) + t.record_sent("r", "a", 0) + t.record_sent("r", "b", 0) + t.resolve_on_relay_accepted("r", "a", 0) # 'a' delivered + assert t.drain_recipient("r", 0) == ["b"] # only 'b' still in flight + + def test_resolve_best_effort_oldest_when_no_id(self) -> None: + # Older relay: MessageSent carries no matching id -> drop oldest. + t = _RecipientInFlightTracker(ttl_ms=10_000) + t.record_sent("r", "a", 0) + t.record_sent("r", "b", 0) + t.resolve_on_relay_accepted("r", None, 0) + assert t.drain_recipient("r", 0) == ["b"] + + def test_ttl_expiry(self) -> None: + t = _RecipientInFlightTracker(ttl_ms=1000) + t.record_sent("r", "a", 0) + assert t.drain_recipient("r", 5000) == [] # older than TTL -> not failed + + def test_prune_and_clear(self) -> None: + t = _RecipientInFlightTracker(ttl_ms=1000) + t.record_sent("r", "a", 0) + t.prune(5000) + assert t.drain_recipient("r", 0) == [] + t.record_sent("r2", "b", 0) + t.clear() + assert t.drain_recipient("r2", 0) == [] + + def test_unrecord_undoes_a_failed_send(self) -> None: + # When `await ws.send` raises, the frame never hit the wire and its + # optimistic entry must be taken back, or a later recipient-keyed + # DeliveryError would false-fail a message that was never sent. + t = _RecipientInFlightTracker(ttl_ms=10_000) + t.record_sent("r", "a", 0) + t.record_sent("r", "b", 0) + t.unrecord("r", "b") # 'b' failed to send + assert t.drain_recipient("r", 0) == ["a"] + + def test_unrecord_removes_only_the_newest_matching_id(self) -> None: + # A retry re-records the same id; unrecording the just-recorded send + # must leave the older same-id entry intact (FIFO retry semantics). + t = _RecipientInFlightTracker(ttl_ms=10_000) + t.record_sent("r", "a", 0) + t.record_sent("r", "a", 1) + t.unrecord("r", "a") + assert t.drain_recipient("r", 0) == ["a"] + + def test_unrecord_missing_is_a_noop(self) -> None: + t = _RecipientInFlightTracker(ttl_ms=10_000) + t.record_sent("r", "a", 0) + t.unrecord("r", "absent") # id not present + t.unrecord("nobody", "a") # recipient not present + assert t.drain_recipient("r", 0) == ["a"] + # dropping the last id for a recipient prunes the empty bucket + t.unrecord("r", "a") + assert t.drain_recipient("r", 0) == [] + + +class TestDeliveryErrorRecipientKeyed: + def test_delivery_error_fails_all_in_flight_for_recipient( + self, mock_protocol: MagicMock + ) -> None: + mgr = InternetManager(mock_protocol, "dev-1") + # two frames in flight to the same recipient + mgr._inflight.record_sent("offX", "id-1", mgr._now_ms()) + mgr._inflight.record_sent("offX", "id-2", mgr._now_ms()) + # relay says recipient offline; older relays send no message_id + mgr._process_received( + json.dumps({ + "type": "DeliveryError", + "recipient": "offX", + "reason": "Recipient is offline", + }).encode() + ) + failed = { + c.kwargs.get("message_id") + for c in mock_protocol.internet_send_failed_with_reason.call_args_list + } + assert failed == {"id-1", "id-2"} + # every fail is tagged with the core's classification prefix + for c in mock_protocol.internet_send_failed_with_reason.call_args_list: + assert c.kwargs.get("reason", "").startswith("recipient_unreachable:") + # and the peer is fed as offline so the core parks + watches for return + mock_protocol.internet_peer_presence.assert_called_once() + assert mock_protocol.internet_peer_presence.call_args.kwargs["peer_id"] == "offX" + assert mock_protocol.internet_peer_presence.call_args.kwargs["online"] is False + + def test_message_sent_prevents_false_fail_of_delivered( + self, mock_protocol: MagicMock + ) -> None: + mgr = InternetManager(mock_protocol, "dev-1") + mgr._inflight.record_sent("offY", "delivered", mgr._now_ms()) + mgr._inflight.record_sent("offY", "stuck", mgr._now_ms()) + # relay confirms 'delivered' was forwarded + mgr._process_received( + json.dumps({ + "type": "MessageSent", + "recipient": "offY", + "message_id": "delivered", + }).encode() + ) + # later DeliveryError for the recipient must NOT fail the delivered one + mgr._process_received( + json.dumps({ + "type": "DeliveryError", + "recipient": "offY", + "reason": "offline", + }).encode() + ) + failed = { + c.kwargs.get("message_id") + for c in mock_protocol.internet_send_failed_with_reason.call_args_list + } + assert "delivered" not in failed + assert "stuck" in failed diff --git a/bindings/react-native/android/src/main/java/com/offlineprotocol/OfflineProtocolModule.kt b/bindings/react-native/android/src/main/java/com/offlineprotocol/OfflineProtocolModule.kt index 96b5bfc3..80c53bf4 100644 --- a/bindings/react-native/android/src/main/java/com/offlineprotocol/OfflineProtocolModule.kt +++ b/bindings/react-native/android/src/main/java/com/offlineprotocol/OfflineProtocolModule.kt @@ -3301,7 +3301,8 @@ class OfflineProtocolModule(reactContext: ReactApplicationContext) : backoffMultiplier = json.optDouble("backoffMultiplier", 2.0).toFloat(), outboxMaxLifetimeMs = json.optLong("outboxMaxLifetimeMs", 604800000).toULong(), pendingMessageMaxLifetimeMs = - json.optLong("pendingMessageMaxLifetimeMs", 604800000).toULong() + json.optLong("pendingMessageMaxLifetimeMs", 604800000).toULong(), + edgeDrivenUnreachableDm = json.optBoolean("edgeDrivenUnreachableDm", false) ) protocol?.updateRetryConfig(retryConfig) diff --git a/bindings/react-native/android/src/main/java/uniffi/offline_protocol/offline_protocol.kt b/bindings/react-native/android/src/main/java/uniffi/offline_protocol/offline_protocol.kt index 364d6fbf..5875b015 100644 --- a/bindings/react-native/android/src/main/java/uniffi/offline_protocol/offline_protocol.kt +++ b/bindings/react-native/android/src/main/java/uniffi/offline_protocol/offline_protocol.kt @@ -8491,6 +8491,8 @@ data class ProtocolConfig ( , var `overflowPolicy`: OverflowPolicy , + var `edgeDrivenUnreachableDm`: kotlin.Boolean = false + , var `maxGroupMembers`: kotlin.UInt = 256u , var `groupRelayEnabled`: kotlin.Boolean = true @@ -8551,6 +8553,7 @@ public object FfiConverterTypeProtocolConfig: FfiConverterRustBuffer { FfiConverterFloat.read(buf), FfiConverterULong.read(buf), FfiConverterULong.read(buf), + FfiConverterBoolean.read(buf), ) } @@ -8908,7 +8916,8 @@ public object FfiConverterTypeRetryConfig: FfiConverterRustBuffer { FfiConverterULong.allocationSize(value.`maxDelayMs`) + FfiConverterFloat.allocationSize(value.`backoffMultiplier`) + FfiConverterULong.allocationSize(value.`outboxMaxLifetimeMs`) + - FfiConverterULong.allocationSize(value.`pendingMessageMaxLifetimeMs`) + FfiConverterULong.allocationSize(value.`pendingMessageMaxLifetimeMs`) + + FfiConverterBoolean.allocationSize(value.`edgeDrivenUnreachableDm`) ) override fun write(value: RetryConfig, buf: ByteBuffer) { @@ -8918,6 +8927,7 @@ public object FfiConverterTypeRetryConfig: FfiConverterRustBuffer { FfiConverterFloat.write(value.`backoffMultiplier`, buf) FfiConverterULong.write(value.`outboxMaxLifetimeMs`, buf) FfiConverterULong.write(value.`pendingMessageMaxLifetimeMs`, buf) + FfiConverterBoolean.write(value.`edgeDrivenUnreachableDm`, buf) } } diff --git a/bindings/react-native/ios/Generated/offline_protocol.swift b/bindings/react-native/ios/Generated/offline_protocol.swift index aa9a25a4..f0911c3e 100644 --- a/bindings/react-native/ios/Generated/offline_protocol.swift +++ b/bindings/react-native/ios/Generated/offline_protocol.swift @@ -5243,6 +5243,7 @@ public struct ProtocolConfig: Equatable, Hashable { public var maxPendingGlobal: UInt64 public var pendingTtlMs: UInt64 public var overflowPolicy: OverflowPolicy + public var edgeDrivenUnreachableDm: Bool public var maxGroupMembers: UInt32 public var groupRelayEnabled: Bool public var groupRelayBroadcastEnabled: Bool @@ -5261,7 +5262,7 @@ public struct ProtocolConfig: Equatable, Hashable { // Default memberwise initializers are never public by default, so we // declare one manually. - public init(appId: String, profile: String, bleEnabled: Bool, wifiDirectEnabled: Bool, internetEnabled: Bool, reticulumEnabled: Bool, nostrEnabled: Bool, preferOnline: Bool, initialTtl: UInt8, encryptionEnabled: Bool, autoKeyExchange: Bool, storePending: Bool, requireEncryption: Bool = true, maxPendingPerPeer: UInt64, maxPendingGlobal: UInt64, pendingTtlMs: UInt64, overflowPolicy: OverflowPolicy, maxGroupMembers: UInt32 = UInt32(256), groupRelayEnabled: Bool = true, groupRelayBroadcastEnabled: Bool = true, groupEnforceAdminCommits: Bool = false, requireTransportIdentity: Bool = false, binaryWireEnabled: Bool = true, nostrSealingEnabled: Bool = true, nostrColdContactEnabled: Bool = true, nostrUsernameDiscoveryEnabled: Bool = false, compactEnvelopeEnabled: Bool = true, richPayloadEnabled: Bool = true, cryptoRecoveryEnabled: Bool = true, meshRelay: MeshRelayConfig? = nil, dataEnabled: Bool = true, controlFreshnessEnforced: Bool = true) { + public init(appId: String, profile: String, bleEnabled: Bool, wifiDirectEnabled: Bool, internetEnabled: Bool, reticulumEnabled: Bool, nostrEnabled: Bool, preferOnline: Bool, initialTtl: UInt8, encryptionEnabled: Bool, autoKeyExchange: Bool, storePending: Bool, requireEncryption: Bool = true, maxPendingPerPeer: UInt64, maxPendingGlobal: UInt64, pendingTtlMs: UInt64, overflowPolicy: OverflowPolicy, edgeDrivenUnreachableDm: Bool = false, maxGroupMembers: UInt32 = UInt32(256), groupRelayEnabled: Bool = true, groupRelayBroadcastEnabled: Bool = true, groupEnforceAdminCommits: Bool = false, requireTransportIdentity: Bool = false, binaryWireEnabled: Bool = true, nostrSealingEnabled: Bool = true, nostrColdContactEnabled: Bool = true, nostrUsernameDiscoveryEnabled: Bool = false, compactEnvelopeEnabled: Bool = true, richPayloadEnabled: Bool = true, cryptoRecoveryEnabled: Bool = true, meshRelay: MeshRelayConfig? = nil, dataEnabled: Bool = true, controlFreshnessEnforced: Bool = true) { self.appId = appId self.profile = profile self.bleEnabled = bleEnabled @@ -5279,6 +5280,7 @@ public struct ProtocolConfig: Equatable, Hashable { self.maxPendingGlobal = maxPendingGlobal self.pendingTtlMs = pendingTtlMs self.overflowPolicy = overflowPolicy + self.edgeDrivenUnreachableDm = edgeDrivenUnreachableDm self.maxGroupMembers = maxGroupMembers self.groupRelayEnabled = groupRelayEnabled self.groupRelayBroadcastEnabled = groupRelayBroadcastEnabled @@ -5327,6 +5329,7 @@ public struct FfiConverterTypeProtocolConfig: FfiConverterRustBuffer { maxPendingGlobal: FfiConverterUInt64.read(from: &buf), pendingTtlMs: FfiConverterUInt64.read(from: &buf), overflowPolicy: FfiConverterTypeOverflowPolicy.read(from: &buf), + edgeDrivenUnreachableDm: FfiConverterBool.read(from: &buf), maxGroupMembers: FfiConverterUInt32.read(from: &buf), groupRelayEnabled: FfiConverterBool.read(from: &buf), groupRelayBroadcastEnabled: FfiConverterBool.read(from: &buf), @@ -5363,6 +5366,7 @@ public struct FfiConverterTypeProtocolConfig: FfiConverterRustBuffer { FfiConverterUInt64.write(value.maxPendingGlobal, into: &buf) FfiConverterUInt64.write(value.pendingTtlMs, into: &buf) FfiConverterTypeOverflowPolicy.write(value.overflowPolicy, into: &buf) + FfiConverterBool.write(value.edgeDrivenUnreachableDm, into: &buf) FfiConverterUInt32.write(value.maxGroupMembers, into: &buf) FfiConverterBool.write(value.groupRelayEnabled, into: &buf) FfiConverterBool.write(value.groupRelayBroadcastEnabled, into: &buf) @@ -5700,16 +5704,18 @@ public struct RetryConfig: Equatable, Hashable { public var backoffMultiplier: Float public var outboxMaxLifetimeMs: UInt64 public var pendingMessageMaxLifetimeMs: UInt64 + public var edgeDrivenUnreachableDm: Bool // Default memberwise initializers are never public by default, so we // declare one manually. - public init(maxRetries: UInt32, initialDelayMs: UInt64, maxDelayMs: UInt64, backoffMultiplier: Float, outboxMaxLifetimeMs: UInt64, pendingMessageMaxLifetimeMs: UInt64) { + public init(maxRetries: UInt32, initialDelayMs: UInt64, maxDelayMs: UInt64, backoffMultiplier: Float, outboxMaxLifetimeMs: UInt64, pendingMessageMaxLifetimeMs: UInt64, edgeDrivenUnreachableDm: Bool = false) { self.maxRetries = maxRetries self.initialDelayMs = initialDelayMs self.maxDelayMs = maxDelayMs self.backoffMultiplier = backoffMultiplier self.outboxMaxLifetimeMs = outboxMaxLifetimeMs self.pendingMessageMaxLifetimeMs = pendingMessageMaxLifetimeMs + self.edgeDrivenUnreachableDm = edgeDrivenUnreachableDm } @@ -5731,7 +5737,8 @@ public struct FfiConverterTypeRetryConfig: FfiConverterRustBuffer { maxDelayMs: FfiConverterUInt64.read(from: &buf), backoffMultiplier: FfiConverterFloat.read(from: &buf), outboxMaxLifetimeMs: FfiConverterUInt64.read(from: &buf), - pendingMessageMaxLifetimeMs: FfiConverterUInt64.read(from: &buf) + pendingMessageMaxLifetimeMs: FfiConverterUInt64.read(from: &buf), + edgeDrivenUnreachableDm: FfiConverterBool.read(from: &buf) ) } @@ -5742,6 +5749,7 @@ public struct FfiConverterTypeRetryConfig: FfiConverterRustBuffer { FfiConverterFloat.write(value.backoffMultiplier, into: &buf) FfiConverterUInt64.write(value.outboxMaxLifetimeMs, into: &buf) FfiConverterUInt64.write(value.pendingMessageMaxLifetimeMs, into: &buf) + FfiConverterBool.write(value.edgeDrivenUnreachableDm, into: &buf) } } diff --git a/bindings/react-native/ios/OfflineProtocolModule.swift b/bindings/react-native/ios/OfflineProtocolModule.swift index 0342abe9..782a7cf4 100644 --- a/bindings/react-native/ios/OfflineProtocolModule.swift +++ b/bindings/react-native/ios/OfflineProtocolModule.swift @@ -3440,7 +3440,8 @@ class OfflineProtocolModule: RCTEventEmitter { maxDelayMs: (config["maxDelayMs"] as? NSNumber)?.uint64Value ?? 300000, backoffMultiplier: (config["backoffMultiplier"] as? NSNumber)?.floatValue ?? 2.0, outboxMaxLifetimeMs: (config["outboxMaxLifetimeMs"] as? NSNumber)?.uint64Value ?? 604800000, - pendingMessageMaxLifetimeMs: (config["pendingMessageMaxLifetimeMs"] as? NSNumber)?.uint64Value ?? 604800000 + pendingMessageMaxLifetimeMs: (config["pendingMessageMaxLifetimeMs"] as? NSNumber)?.uint64Value ?? 604800000, + edgeDrivenUnreachableDm: (config["edgeDrivenUnreachableDm"] as? NSNumber)?.boolValue ?? false ) try proto.updateRetryConfig(config: retryConfig) diff --git a/bindings/react-native/src/types.ts b/bindings/react-native/src/types.ts index 7e0844fb..dff0272a 100644 --- a/bindings/react-native/src/types.ts +++ b/bindings/react-native/src/types.ts @@ -65,6 +65,15 @@ export interface RetryConfig { * terminal `message_failed` event is emitted (default 7 days). */ pendingMessageMaxLifetimeMs?: number; + /** + * Opt-in (default false): treat a durably-unreachable direct message as + * edge-driven after a few reachability probes instead of probing it forever + * (15s→600s cap). Zeros steady-state relay traffic to gone peers and makes a + * restart skip re-driving a durably-failing backlog, at the cost of the + * timed-probe self-recovery guarantee for a silent returning peer. Safe only + * where peers always interact or advertise presence on return (e.g. M2M). + */ + edgeDrivenUnreachableDm?: boolean; } export interface DedupConfig { diff --git a/crates/offline-protocol-reliability/src/ack_manager.rs b/crates/offline-protocol-reliability/src/ack_manager.rs index 7024948b..ac6ab86d 100644 --- a/crates/offline-protocol-reliability/src/ack_manager.rs +++ b/crates/offline-protocol-reliability/src/ack_manager.rs @@ -264,6 +264,22 @@ impl AckManager { } } + /// Carry an accumulated retry count onto a pending ACK. + /// + /// `register_pending_ack*` always starts a fresh ACK at `retry_count: 0`. + /// When a message is *re-sent* from the retry queue and its prior ACK had + /// already been cleared (the resend registers a brand-new ACK), that new ACK + /// would otherwise reset the backoff ladder to 0 — pinning `delay_for_retry` + /// at its 1s floor for a recipient that never ACKs (an offline peer), which + /// re-sends ~once per second forever. Seeding the new ACK with the count the + /// retry-queue entry already carries lets the exponential backoff advance + /// across resends. No-op if the message has no pending ACK. + pub fn set_retry_count(&mut self, message_id: &MessageId, retry_count: u32) { + if let Some(pending) = self.pending_acks.get_mut(message_id) { + pending.retry_count = retry_count; + } + } + /// Gets information about a pending ACK. pub fn get_pending_ack(&self, message_id: &MessageId) -> Option<&PendingAck> { self.pending_acks.get(message_id) @@ -366,6 +382,34 @@ mod tests { assert_eq!(pending.retry_count, 1); } + #[test] + fn test_set_retry_count_carries_backoff_forward() { + // Pins the flap fix: when a relay-accepted resend registers a FRESH ACK + // (the prior one was cleared by an ACK timeout that re-queued the + // message), the backoff ladder must be seeded from the accumulated + // count instead of resetting to 0. Resetting to 0 pins delay_for_retry + // to its 1s floor forever for an offline peer, which floods the relay + // and flaps the connection. If a future change reverts set_retry_count + // to a no-op or drops the call site, this test fails loudly. + let mut manager = AckManager::new(); + let msg_id = MessageId::new(); + + // Fresh ACK always starts at 0 (register hardcodes it). + manager.register_pending_ack(msg_id.clone(), None).unwrap(); + assert_eq!(manager.get_pending_ack(&msg_id).unwrap().retry_count, 0); + + // Carry an accumulated count forward onto the freshly-registered ACK. + manager.set_retry_count(&msg_id, 5); + assert_eq!(manager.get_pending_ack(&msg_id).unwrap().retry_count, 5); + + // Idempotent/absolute set (not increment). + manager.set_retry_count(&msg_id, 7); + assert_eq!(manager.get_pending_ack(&msg_id).unwrap().retry_count, 7); + + // No-op for an unknown message (no pending ACK). + manager.set_retry_count(&MessageId::new(), 9); + } + #[test] fn test_remove_ack() { let mut manager = AckManager::new(); diff --git a/crates/offline-protocol-reliability/src/retry_queue.rs b/crates/offline-protocol-reliability/src/retry_queue.rs index 4c821de2..402d1987 100644 --- a/crates/offline-protocol-reliability/src/retry_queue.rs +++ b/crates/offline-protocol-reliability/src/retry_queue.rs @@ -29,6 +29,26 @@ pub struct RetryConfig { /// Maximum lifetime for messages waiting on MLS session establishment. pub pending_message_max_lifetime_ms: u64, + + /// Opt-in: treat a durably-unreachable direct message as EDGE-DRIVEN once + /// its escalating reachability probe has run a few times, instead of the + /// default of probing it forever at the 15s->600s cap on every carrier. + /// + /// Default `false` preserves the documented contract (a parked message + /// "never goes fully quiet"; see docs/message-delivery.md) and every + /// existing native/third-party app's behaviour is unchanged. When `true`, + /// after a bounded number of probes the message stops being timed-probed + /// and rests in the outbox, re-driven only when the peer next proves + /// reachable (an inbound frame or presence-online edge flushes it). It also + /// lets a restart skip re-driving a durably-failing backlog. + /// + /// This trades the SDK's "self-recovers even for a silent returning peer + /// with no presence" guarantee for zero steady-state relay traffic to gone + /// peers. It is correct for deployments whose peers always interact or + /// advertise presence on return (e.g. a machine-to-machine capability + /// exchange), and MUST NOT be enabled for consumers that rely on the timed + /// probe as their only recovery path. + pub edge_driven_unreachable_dm: bool, } impl Default for RetryConfig { @@ -40,6 +60,8 @@ impl Default for RetryConfig { backoff_multiplier: DEFAULT_BACKOFF_MULTIPLIER, outbox_max_lifetime_ms: DEFAULT_OUTBOX_LIFETIME_MS, pending_message_max_lifetime_ms: DEFAULT_PENDING_MESSAGE_LIFETIME_MS, + // Default off: preserve the documented perpetual-probe contract. + edge_driven_unreachable_dm: false, } } } @@ -374,6 +396,16 @@ mod tests { .build() } + #[test] + fn test_edge_driven_unreachable_dm_defaults_off() { + // The opt-in flood-control behaviors (edge-driven parking, restart + // age-gate, core resend rate-cap) are all gated on this flag. It MUST + // default false so every existing native/third-party app keeps the + // documented perpetual-probe behavior unchanged. If a future change + // flips the default, this fails and flags the silent regression. + assert!(!RetryConfig::default().edge_driven_unreachable_dm); + } + #[test] fn test_enqueue_and_dequeue() { let config = RetryConfig { diff --git a/crates/offline-protocol-uniffi/src/lib.rs b/crates/offline-protocol-uniffi/src/lib.rs index d06b38a6..dbf39744 100644 --- a/crates/offline-protocol-uniffi/src/lib.rs +++ b/crates/offline-protocol-uniffi/src/lib.rs @@ -2169,6 +2169,9 @@ pub struct RetryConfig { pub backoff_multiplier: f32, pub outbox_max_lifetime_ms: u64, pub pending_message_max_lifetime_ms: u64, + /// Opt-in (default false): edge-driven delivery for durably-unreachable DMs. + /// See `RetryConfig::edge_driven_unreachable_dm` and the UDL dictionary. + pub edge_driven_unreachable_dm: bool, } /// Deduplication configuration @@ -2318,6 +2321,15 @@ pub struct ProtocolConfig { pub max_pending_global: u64, pub pending_ttl_ms: u64, pub overflow_policy: OverflowPolicy, + /// Opt-in (default false): treat a durably-unreachable direct message as + /// edge-driven after a few reachability probes, instead of probing it + /// forever at the 15s->600s cap. Zeros steady-state relay traffic to gone + /// peers and makes a restart skip re-driving a durably-failing backlog, at + /// the cost of the timed-probe self-recovery guarantee for a silent + /// returning peer. See the UDL dictionary and + /// `RetryConfig::edge_driven_unreachable_dm`. Safe only where peers always + /// interact or advertise presence on return (e.g. machine-to-machine). + pub edge_driven_unreachable_dm: bool, pub max_group_members: u32, pub group_relay_enabled: bool, /// Whether a relay-synced group may send one O(1) relay broadcast instead @@ -2619,6 +2631,8 @@ impl From for CoreConfig { core_config.encryption.enabled = config.encryption_enabled; core_config.encryption.auto_key_exchange = config.auto_key_exchange; core_config.encryption.store_pending = config.store_pending; + core_config.reliability.retry.edge_driven_unreachable_dm = + config.edge_driven_unreachable_dm; core_config.encryption.require_encryption = config.require_encryption; core_config.encryption.pending_queue = CorePendingQueueConfig { max_pending_per_peer: config.max_pending_per_peer as usize, @@ -6162,6 +6176,7 @@ impl OfflineProtocol { backoff_multiplier: config.backoff_multiplier, outbox_max_lifetime_ms: config.outbox_max_lifetime_ms, pending_message_max_lifetime_ms: config.pending_message_max_lifetime_ms, + edge_driven_unreachable_dm: config.edge_driven_unreachable_dm, }; let mut protocol = self.lock_inner_recovering(); protocol @@ -7946,6 +7961,7 @@ mod tests { max_pending_global: 4096, pending_ttl_ms: DEFAULT_PENDING_TTL_MS, overflow_policy: OverflowPolicy::DropOldest, + edge_driven_unreachable_dm: false, max_group_members: 256, group_relay_enabled: true, group_relay_broadcast_enabled: true, @@ -7983,6 +7999,7 @@ mod tests { max_pending_global: 4096, pending_ttl_ms: DEFAULT_PENDING_TTL_MS, overflow_policy: OverflowPolicy::DropOldest, + edge_driven_unreachable_dm: false, max_group_members: 256, group_relay_enabled: true, group_relay_broadcast_enabled: true, @@ -8297,6 +8314,7 @@ mod tests { max_pending_global: 4096, pending_ttl_ms: DEFAULT_PENDING_TTL_MS, overflow_policy: OverflowPolicy::DropOldest, + edge_driven_unreachable_dm: false, max_group_members: 256, group_relay_enabled: true, group_relay_broadcast_enabled: true, diff --git a/crates/offline-protocol-uniffi/src/offline_protocol.udl b/crates/offline-protocol-uniffi/src/offline_protocol.udl index 7bc97379..47567cbe 100644 --- a/crates/offline-protocol-uniffi/src/offline_protocol.udl +++ b/crates/offline-protocol-uniffi/src/offline_protocol.udl @@ -610,6 +610,9 @@ dictionary RetryConfig { f32 backoff_multiplier; u64 outbox_max_lifetime_ms; u64 pending_message_max_lifetime_ms; + // Opt-in (default false): edge-driven delivery for durably-unreachable DMs. + // See RetryConfig::edge_driven_unreachable_dm. + boolean edge_driven_unreachable_dm = false; }; // Deduplication configuration @@ -701,6 +704,14 @@ dictionary ProtocolConfig { u64 max_pending_global; u64 pending_ttl_ms; OverflowPolicy overflow_policy; + // Opt-in (default false): treat a durably-unreachable direct message as + // edge-driven after a few reachability probes instead of probing it forever + // (15s->600s cap). Zeros steady-state relay traffic to gone peers and makes + // a restart skip re-driving a durably-failing backlog, at the cost of the + // timed-probe self-recovery guarantee for a silent returning peer. Safe only + // where peers always interact or advertise presence on return (e.g. M2M). + // See RetryConfig::edge_driven_unreachable_dm. + boolean edge_driven_unreachable_dm = false; u32 max_group_members = 256; boolean group_relay_enabled = true; // Whether a relay-synced group may send one O(1) relay broadcast instead diff --git a/crates/offline-protocol/src/protocol/mod.rs b/crates/offline-protocol/src/protocol/mod.rs index e6cb3ce0..780b7070 100644 --- a/crates/offline-protocol/src/protocol/mod.rs +++ b/crates/offline-protocol/src/protocol/mod.rs @@ -559,6 +559,26 @@ pub struct OfflineProtocol { /// Probe schedule for pending sessions to guarantee post-restart convergence. confirmation_probe_due_at: HashMap>, + /// Consecutive relay "recipient unreachable" verdicts per peer for the + /// session-confirmation probe. Drives an escalating backoff of + /// `confirmation_probe_due_at` (15s -> 600s cap, the same ladder the welcome + /// lifecycle uses) so a peer that established a session then vanished is not + /// probed every `CONFIRMATION_PROBE_INTERVAL_SECS` forever; a fleet of such + /// peers would otherwise exceed the relay's per-connection rate limit and + /// flap the connection. Reset on a reachability edge (`on_peer_presence`). + confirmation_probe_unreachable_parks: HashMap, + + /// Token bucket that caps how fast the RETRY/PROBE path + /// (`process_retry_queue`) re-sends, so a large backlog of resends to + /// unreachable peers cannot burst past the relay's per-connection rate limit + /// and trigger the disconnect->reconnect->reflood loop ("flapping"). Only + /// consulted when `RetryConfig::edge_driven_unreachable_dm` is opted in. It + /// bounds ONLY resends — first-attempt sends, ACKs, control frames and + /// handshakes take other paths and are never throttled, so live traffic + /// keeps full throughput/latency. `retry_drain_last` is None until first use. + retry_drain_tokens: f64, + retry_drain_last: Option, + /// Rate-limit schedule for 1:1 session re-keys triggered by an epoch-desync /// decrypt failure. Bounds the re-key to at most one per peer per /// `REKEY_INTERVAL_SECS` so a peer replaying stale-epoch ciphertext (or an @@ -973,6 +993,9 @@ impl OfflineProtocol { lamport_clock: LamportClock::new(), confirmation_retry_due_at: HashMap::new(), confirmation_probe_due_at: HashMap::new(), + confirmation_probe_unreachable_parks: HashMap::new(), + retry_drain_tokens: 0.0, + retry_drain_last: None, rekey_due_at: HashMap::new(), welcome_lifecycles: HashMap::new(), pending_connection_requests: HashMap::new(), @@ -2151,6 +2174,33 @@ impl OfflineProtocol { } } + // Opt-in (`edge_driven_unreachable_dm`): on the start/reconnect re-drive, + // leave durably-failing messages (first sent longer ago than + // OUTBOX_DURABLE_UNREACHABLE_AGE_SECS, still undelivered) edge-driven + // rather than re-driving them, so a restart does not re-ramp a large + // dead-peer backlog. They rest in the outbox and flush the instant their + // peer next proves reachable. Default-off preserves the documented + // "reconnect re-drives, ignoring timing" behaviour for every existing + // app. drain_all() already removed these from the retry queue and the + // outbox entry is untouched, so "not collected" == "resting edge-driven". + if self.config.reliability.retry.edge_driven_unreachable_dm { + let now = Utc::now(); + let durable_age = chrono::Duration::seconds( + crate::protocol::types::OUTBOX_DURABLE_UNREACHABLE_AGE_SECS, + ); + all_messages.retain(|(m, _)| { + match self + .outbox + .get(&m.id) + .or_else(|| self.media_outbox.get(&m.id)) + .map(|e| e.first_sent_at) + { + Some(first_sent) => now.signed_duration_since(first_sent) < durable_age, + None => true, + } + }); + } + if all_messages.is_empty() { return; } @@ -3186,7 +3236,35 @@ impl OfflineProtocol { /// - Properly tracks retry counts and transport failures fn process_retry_queue(&mut self) -> Result<()> { // Limit batch size to prevent blocking on large queues - let max_batch_size = crate::constants::FLUSH_BATCH_LIMIT; + let mut max_batch_size = crate::constants::FLUSH_BATCH_LIMIT; + + // Opt-in resend rate cap: bound how fast the RETRY/PROBE path re-sends, + // so a backlog of resends to unreachable peers cannot burst past the + // relay's per-connection rate limit and flap the connection. Refills a + // token bucket by elapsed wall-time and caps this call's batch to the + // whole tokens available. Only resends are bounded; first-attempt sends, + // ACKs, control frames and handshakes take other paths and are never + // throttled, so live traffic keeps full throughput and latency. Fully + // off by default (unchanged retry behaviour for every existing app). + if self.config.reliability.retry.edge_driven_unreachable_dm { + let now = Instant::now(); + match self.retry_drain_last { + Some(last) => { + let elapsed = now.duration_since(last).as_secs_f64(); + self.retry_drain_tokens = (self.retry_drain_tokens + + elapsed * crate::protocol::types::RETRY_DRAIN_RATE_PER_SEC) + .min(crate::protocol::types::RETRY_DRAIN_BURST); + } + None => { + // First drain: start with a full burst budget. + self.retry_drain_tokens = crate::protocol::types::RETRY_DRAIN_BURST; + } + } + self.retry_drain_last = Some(now); + let budget = self.retry_drain_tokens.floor() as usize; + max_batch_size = max_batch_size.min(budget); + } + let mut processed = 0; while processed < max_batch_size { @@ -3218,7 +3296,21 @@ impl OfflineProtocol { Ok(()) => { let ack_registered_now = self.ensure_ack_registration(&entry.message)?; - if !ack_registered_now { + if ack_registered_now { + // The resend registered a brand-new ACK (the prior one + // was cleared, e.g. by an ACK timeout that re-queued + // this message). A fresh ACK starts at retry_count 0, so + // carry the count the retry-queue entry already holds — + // advanced by this attempt — onto it. Without this the + // backoff ladder resets to 0 on every relay-accepted + // resend and `delay_for_retry` is pinned at its 1s floor + // forever for a peer that never ACKs (an offline + // recipient), flooding the relay ~once per second. + self.ack_manager.set_retry_count( + &entry.message.id, + entry.retry_count.saturating_add(1), + ); + } else { self.ack_manager.increment_retry_count(&entry.message.id); } self.mark_message_sent( @@ -3273,6 +3365,10 @@ impl OfflineProtocol { if processed > 0 { debug!(processed = processed, "Processed retry queue entries"); + // Spend the resend-rate tokens actually used (opt-in cap only). + if self.config.reliability.retry.edge_driven_unreachable_dm { + self.retry_drain_tokens = (self.retry_drain_tokens - processed as f64).max(0.0); + } } Ok(()) diff --git a/crates/offline-protocol/src/protocol/send.rs b/crates/offline-protocol/src/protocol/send.rs index 9b350154..af69572c 100644 --- a/crates/offline-protocol/src/protocol/send.rs +++ b/crates/offline-protocol/src/protocol/send.rs @@ -3788,6 +3788,12 @@ impl OfflineProtocol { .outbound_media_chunks .get(&parsed_id) .map(|(file_id, _)| file_id.clone()); + // If this peer is also an unconfirmed-session probe target, fold the + // verdict into the probe schedule so it backs off (15s -> 600s) instead + // of re-probing every 5s forever. No-op for peers that are not probe + // targets (e.g. an ordinary DM to a confirmed peer). Done here, after + // the `entry` borrow of `self.outbox` has ended. + self.note_confirmation_probe_unreachable(&recipient); warn!( message_id = %message_id, file_id = ?file_id, @@ -3867,10 +3873,20 @@ impl OfflineProtocol { /// internet-only fleet is on this path: /// - verdict branch: the escalation is the bound — one frame per interval /// per parked *message*, settling at one per 600s; - /// - accepted branch: the probe registers a fresh ACK at `retry_count` 0, - /// so it rides the ordinary ACK ladder (up to `max_retries` sends, - /// 1s → 300s backoff, ~800s cumulative on the defaults) before - /// `try_repark_exhausted_dm` re-parks it at the escalated interval. + /// - accepted branch: the probe registers a fresh ACK, and + /// [`crate::protocol::OfflineProtocol::process_retry_queue`] seeds that + /// ACK with the retry-queue entry's carried count (`retry_count + 1`) + /// rather than letting it restart at 0. A probe whose entry is fresh (a + /// reachability edge re-drove it with a new budget) still walks the full + /// ladder — up to `max_retries` sends, 1s → 300s backoff, ~800s + /// cumulative on the defaults — but a probe whose entry already carries + /// backoff resumes near its current position and re-parks after roughly + /// one more timeout instead of replaying the 1s floor. Plan relay + /// capacity against the fresh-entry number; the carried case only sends + /// less. Either way `try_repark_exhausted_dm` re-parks it at the + /// escalated interval. The carry-forward is load-bearing: without it a + /// never-ACKing (offline) recipient's probe pins `delay_for_retry` at + /// its 1s floor and floods the relay ~once per second. /// /// The outbox lifetime bounds the entry itself — and note the probe /// refreshes `last_sent_at` on every send, so the sliding 7-day window @@ -3927,11 +3943,23 @@ impl OfflineProtocol { // bounds how often we ask, and each offer spends the same own-send // tokens as any other frame of ours. handed_to_mesh = self.offer_to_mesh(&message); - let _ = self.retry_queue.enqueue_with_delay( - message, - attempt_count, - (retry_in_secs * 1000) as u64, - ); + // Default: keep the documented perpetual timed probe (15s->600s cap + // on every carrier). Opt-in (`edge_driven_unreachable_dm`): after a + // bounded number of probes, stop the timer and leave the message in + // the outbox edge-driven — re-driven only when the peer next proves + // reachable (inbound frame / presence-online -> on_neighbor_discovered + // -> flush_outbox_for_peer_via). This zeroes steady-state relay + // traffic to a durably-gone peer, at the cost of the timed-probe + // self-recovery guarantee for a silent returning peer. + let edge_driven = self.config.reliability.retry.edge_driven_unreachable_dm + && parks > crate::protocol::types::DM_UNREACHABLE_PROBE_LIMIT; + if !edge_driven { + let _ = self.retry_queue.enqueue_with_delay( + message, + attempt_count, + (retry_in_secs * 1000) as u64, + ); + } } debug!( message_id = %message_id, diff --git a/crates/offline-protocol/src/protocol/session.rs b/crates/offline-protocol/src/protocol/session.rs index e499df26..5aa02761 100644 --- a/crates/offline-protocol/src/protocol/session.rs +++ b/crates/offline-protocol/src/protocol/session.rs @@ -447,6 +447,45 @@ impl OfflineProtocol { pub(super) fn clear_confirmation_recovery_tracking(&mut self, peer_id: &str) { self.confirmation_retry_due_at.remove(peer_id); self.confirmation_probe_due_at.remove(peer_id); + self.confirmation_probe_unreachable_parks.remove(peer_id); + } + + /// Fold a relay "recipient unreachable" verdict into the session-confirmation + /// probe schedule: escalate this peer's next probe on the same 15s -> 600s + /// ladder the welcome lifecycle uses (see + /// [`Self::apply_recipient_unreachable_failure`]). Without this, a peer that + /// established an MLS session then vanished before confirming is re-probed + /// every `CONFIRMATION_PROBE_INTERVAL_SECS` (5s) indefinitely, because + /// `kick_pending_session_reconciliation` re-arms it every scan and + /// `on_transport_send_failed` otherwise never consults this scheduler. A + /// handful of such peers pushes aggregate relay traffic past the + /// per-connection rate limit, which disconnects the socket on a loop. + /// + /// No-op unless the peer is currently a probe target (a `confirmation_probe` + /// entry exists, so a probe was just attempted), so ordinary DM failures to + /// confirmed peers do not touch the probe schedule. The counter is reset on + /// a reachability edge in [`Self::on_peer_presence`] so a returning peer is + /// re-probed immediately and its session still converges. + pub(super) fn note_confirmation_probe_unreachable(&mut self, peer_id: &str) { + if !self.confirmation_probe_due_at.contains_key(peer_id) { + return; + } + let parks = { + let counter = self + .confirmation_probe_unreachable_parks + .entry(peer_id.to_string()) + .or_insert(0); + *counter = counter.saturating_add(1); + *counter + }; + // parks >= 1; shift clamped so 15 << 6 = 960 is the largest pre-cap + // value (no overflow), then capped — identical to the welcome ladder. + let backoff_secs = (WELCOME_NO_CARRIER_RETRY_SECS << (parks - 1).min(6)) + .min(WELCOME_UNREACHABLE_RETRY_CAP_SECS); + self.confirmation_probe_due_at.insert( + peer_id.to_string(), + Utc::now() + ChronoDuration::seconds(backoff_secs), + ); } pub(super) fn collect_pending_session_peers(&mut self) -> Result> { @@ -546,6 +585,8 @@ impl OfflineProtocol { let pending_set: HashSet = pending_peers.iter().cloned().collect(); self.confirmation_probe_due_at .retain(|peer, _| pending_set.contains(peer)); + self.confirmation_probe_unreachable_parks + .retain(|peer, _| pending_set.contains(peer)); for peer_id in pending_peers { let due_at = self @@ -1324,6 +1365,16 @@ impl OfflineProtocol { self.resend_unconfirmed_sent_welcome(peer_id, "peer_presence_online"); self.note_welcome_rescue_attempt(peer_id); } + // A reachability edge resets the session-confirmation probe's + // unreachable backoff: clear the escalation counter and mark a probe + // due now, so a peer that went unreachable while unconfirmed is + // re-probed immediately and its session converges rather than + // waiting out the (up to 600s) backoff. + if self.confirmation_probe_due_at.contains_key(peer_id) { + self.confirmation_probe_unreachable_parks.remove(peer_id); + self.confirmation_probe_due_at + .insert(peer_id.to_string(), Utc::now()); + } } else { self.park_welcome_peer_unreachable(peer_id); } diff --git a/crates/offline-protocol/src/protocol/tests/mod.rs b/crates/offline-protocol/src/protocol/tests/mod.rs index 8c49f64d..c70ec973 100644 --- a/crates/offline-protocol/src/protocol/tests/mod.rs +++ b/crates/offline-protocol/src/protocol/tests/mod.rs @@ -37462,3 +37462,215 @@ fn the_desync_path_still_claims_the_shared_rekey_window() { "a desync-driven re-key did not claim the window that bounds a re-key storm" ); } + +/// Pins component (B) of the flap fix: the confirmation-probe unreachable +/// escalation. Without it an unconfirmed-but-vanished peer is re-probed every +/// 5s forever, and a handful of such peers push aggregate relay traffic past +/// the per-connection rate limit into a disconnect loop. The always-on +/// behavior (never got a regression test in the earlier rounds) is: no-op +/// unless the peer is an active probe target, escalate the shared 15s->600s +/// ladder once per unreachable park, and reset to "probe now" on a +/// reachability edge so a returning peer still converges. +#[test] +fn test_confirmation_probe_unreachable_escalation() { + let mut protocol = OfflineProtocol::new(create_test_config()).unwrap(); + + // No-op gate: a peer that is NOT an active probe target (no + // confirmation_probe_due_at entry) must not touch the probe schedule, so + // ordinary DM failures to confirmed peers never escalate anything. + protocol.note_confirmation_probe_unreachable("ghost"); + assert!( + !protocol + .confirmation_probe_unreachable_parks + .contains_key("ghost"), + "escalation must no-op for a peer that is not an active probe target" + ); + assert!( + !protocol.confirmation_probe_due_at.contains_key("ghost"), + "escalation must not create a probe schedule for a non-target peer" + ); + + // Make "bob" an active probe target (a probe was just attempted). + protocol + .confirmation_probe_due_at + .insert("bob".to_string(), Utc::now()); + + // First unreachable park: counter -> 1, due_at pushed out ~15s (15 << 0). + protocol.note_confirmation_probe_unreachable("bob"); + assert_eq!( + protocol.confirmation_probe_unreachable_parks.get("bob"), + Some(&1), + "first unreachable park must set the escalation counter to 1" + ); + let secs_1 = + (*protocol.confirmation_probe_due_at.get("bob").unwrap() - Utc::now()).num_seconds(); + assert!( + (WELCOME_NO_CARRIER_RETRY_SECS - 2..=WELCOME_NO_CARRIER_RETRY_SECS + 1).contains(&secs_1), + "first park must schedule the probe ~15s out, got {secs_1}s" + ); + + // Second unreachable park: counter -> 2, due_at ~30s (15 << 1). Proves the + // ladder doubles rather than re-probing at the flat 5s interval. + protocol.note_confirmation_probe_unreachable("bob"); + assert_eq!( + protocol.confirmation_probe_unreachable_parks.get("bob"), + Some(&2), + "second unreachable park must escalate the counter to 2" + ); + let secs_2 = + (*protocol.confirmation_probe_due_at.get("bob").unwrap() - Utc::now()).num_seconds(); + assert!( + (2 * WELCOME_NO_CARRIER_RETRY_SECS - 2..=2 * WELCOME_NO_CARRIER_RETRY_SECS + 1) + .contains(&secs_2), + "second park must double the interval to ~30s, got {secs_2}s" + ); + + // A reachability edge (presence online) resets the escalation and marks a + // probe due now, so a peer that vanished while unconfirmed is re-probed + // immediately and its session converges instead of waiting out the backoff. + protocol.on_peer_presence("bob", true, None); + assert!( + !protocol + .confirmation_probe_unreachable_parks + .contains_key("bob"), + "a presence-online edge must clear the escalation counter" + ); + let secs_reset = + (*protocol.confirmation_probe_due_at.get("bob").unwrap() - Utc::now()).num_seconds(); + assert!( + secs_reset <= 2, + "a presence-online edge must mark the probe due now, got {secs_reset}s" + ); +} + +/// Pins component (D) of the flap fix: the opt-in edge-driven parking gate. +/// With `edge_driven_unreachable_dm` on, a durably-unreachable DM keeps being +/// timed-probed up to `DM_UNREACHABLE_PROBE_LIMIT` times, then stops (rests in +/// the outbox, re-driven only on a reachability edge) instead of probing +/// forever. Off (default) it probes on every park — covered by the existing +/// `..._parked_dm_probe_escalates_...` test, which is the control case. +#[test] +fn test_edge_driven_parking_stops_probe_after_limit() { + let mut config = create_test_config(); + config.reliability.retry.edge_driven_unreachable_dm = true; + let mut protocol = OfflineProtocol::new(config).unwrap(); + + let mock_transport = MockTransport::new(TransportType::Internet); + mock_transport.start().unwrap(); + protocol + .transport_manager_mut() + .add_transport(TransportType::Internet, Box::new(mock_transport)); + protocol.start().unwrap(); + + let message_id = protocol + .send_message("bob", "hello", None::, None::) + .unwrap(); + + // Park #1 (== DM_UNREACHABLE_PROBE_LIMIT, not yet over it): still probes, + // so a timed re-check is scheduled exactly as in the default path. + protocol + .on_transport_send_failed( + &message_id.as_str(), + Some("recipient_unreachable: peer offline".to_string()), + ) + .unwrap(); + assert_eq!(protocol.dm_unreachable_parks.get("bob"), Some(&1)); + assert!( + protocol.retry_queue.time_until_next_retry().is_some(), + "within the probe limit, the message must still be timed-probed" + ); + + // Park #2 (> DM_UNREACHABLE_PROBE_LIMIT): goes edge-driven. No timed probe + // is scheduled; the message rests in the outbox to be re-driven on the + // peer's next reachability edge (inbound frame / presence-online). + protocol + .on_transport_send_failed( + &message_id.as_str(), + Some("recipient_unreachable: peer offline".to_string()), + ) + .unwrap(); + assert_eq!(protocol.dm_unreachable_parks.get("bob"), Some(&2)); + assert!( + protocol.retry_queue.time_until_next_retry().is_none(), + "past the probe limit, edge-driven parking must stop the timed probe" + ); + assert!( + protocol.outbox.contains_key(&message_id), + "an edge-driven parked message must stay in the outbox for edge re-drive" + ); + + // The reachability edge still works: presence-online resets the escalation + // and re-drives, so an edge-driven message is never stranded. + protocol.on_peer_presence("bob", true, None); + assert_eq!( + protocol.dm_unreachable_parks.get("bob"), + None, + "a presence-online edge must reset the edge-driven park escalation" + ); +} + +/// Pins component (E) of the flap fix: the opt-in resend rate cap. With +/// `edge_driven_unreachable_dm` on, `process_retry_queue` must bound resends to +/// the token-bucket burst so a backlog of resends to unreachable peers cannot +/// burst past the relay's per-connection rate limit and flap the connection. +/// Off (default) the batch is bounded only by `FLUSH_BATCH_LIMIT` per call and +/// the whole backlog drains — proving the cap is what throttles resends. +#[test] +fn test_resend_rate_cap_bounds_retry_drain_to_burst() { + fn drain_count(edge_driven: bool) -> usize { + let mut config = create_test_config(); + config.reliability.retry.edge_driven_unreachable_dm = edge_driven; + let mut protocol = OfflineProtocol::new(config).unwrap(); + + // A transport that fails every send: each processed entry re-enqueues + // with backoff (not immediately ready) and emits one MessageRetrying, + // so counting those events counts entries processed per call. + let flaky = FlakyTransport::fail_first(TransportType::Internet, u32::MAX); + protocol + .transport_manager_mut() + .add_transport(TransportType::Internet, Box::new(flaky)); + protocol.start().unwrap(); + + let count = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let count_handle = Arc::clone(&count); + protocol.on_event(move |event| { + if matches!(event, Event::MessageRetrying { .. }) { + count_handle.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + } + }); + + // 60 distinct messages, all ready now. + for i in 0..60 { + let msg = unsigned_frame("user123", "bob", format!("m{i}")); + protocol.retry_queue.enqueue_with_delay(msg, 0, 0); + } + + // Three back-to-back drains. Elapsed between them is ~0, so the bucket + // barely refills; the cap (burst 24) binds after the first full batch. + for _ in 0..3 { + protocol.process_retry_queue().unwrap(); + } + count.load(std::sync::atomic::Ordering::SeqCst) + } + + let capped = drain_count(true); + let uncapped = drain_count(false); + assert!( + capped <= crate::protocol::types::RETRY_DRAIN_BURST as usize, + "the resend rate cap must bound a burst of resends to the token-bucket \ + burst ({}), processed {capped}", + crate::protocol::types::RETRY_DRAIN_BURST as usize + ); + assert!( + capped >= crate::constants::FLUSH_BATCH_LIMIT, + "the first drain should still deliver a full batch, processed {capped}" + ); + assert_eq!( + uncapped, 60, + "with the cap off the whole backlog drains, processed {uncapped}" + ); + assert!( + capped < uncapped, + "the cap must throttle resends relative to the default ({capped} vs {uncapped})" + ); +} diff --git a/crates/offline-protocol/src/protocol/types.rs b/crates/offline-protocol/src/protocol/types.rs index c788bddc..4bc09ec2 100644 --- a/crates/offline-protocol/src/protocol/types.rs +++ b/crates/offline-protocol/src/protocol/types.rs @@ -72,6 +72,31 @@ pub(crate) const WELCOME_NO_CARRIER_RETRY_SECS: i64 = 15; /// escalation carries the whole bound, since that probe runs on every carrier /// (see `park_unreachable_dm`), including internet-only devices. pub(crate) const WELCOME_UNREACHABLE_RETRY_CAP_SECS: i64 = 600; +/// Only used when `RetryConfig::edge_driven_unreachable_dm` is opted in. How +/// many escalating timed probes a plain DM to an unreachable peer gets before +/// it goes edge-driven (stops being timed-probed and rests in the outbox until +/// the peer next proves reachable). One probe (plus the original send) gives a +/// brief blip tolerance; beyond that, recovery is the reachability edge — which +/// in an opted-in deployment is reliable (peers interact / advertise presence +/// on return), so more blind probes only lengthen the transient after a mass +/// disconnect. Default-off deployments never consult this (they probe forever +/// at the documented 15s->600s cap). +pub(crate) const DM_UNREACHABLE_PROBE_LIMIT: u32 = 1; +/// Only used when `RetryConfig::edge_driven_unreachable_dm` is opted in. On the +/// startup/reconnect re-drive, a still-undelivered outbox message first sent +/// longer ago than this is left edge-driven rather than re-driven, so a restart +/// does not re-ramp a durably-failing backlog. Default-off deployments never +/// consult this. +pub(crate) const OUTBOX_DURABLE_UNREACHABLE_AGE_SECS: i64 = 300; +/// Only used when `RetryConfig::edge_driven_unreachable_dm` is opted in. +/// Sustained rate (per second) and burst budget for the RETRY/PROBE resend path +/// (`process_retry_queue`). Set below the reference relay's per-connection limit +/// (10/s sustained, 30 burst) so resends to a fleet of unreachable peers can +/// never burst the connection into a rate-limit disconnect. Only resends are +/// bounded; first-attempt sends, ACKs, control frames and handshakes are never +/// throttled. +pub(crate) const RETRY_DRAIN_RATE_PER_SEC: f64 = 8.0; +pub(crate) const RETRY_DRAIN_BURST: f64 = 24.0; /// Age limit for a welcome lifecycle to keep its peer on the presence /// watchlist (`welcome_pending_peers`). Without it the watch set only ever /// grows: every offline presence answer re-parks the record and pushes its diff --git a/docs/configuration.md b/docs/configuration.md index 03fbcd66..f593080c 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -521,6 +521,28 @@ read them. | `backoffMultiplier` | number | 2.0 | Backoff multiplier | | `outboxMaxLifetimeMs` | number | 604800000 | Max message lifetime (7 days) | | `pendingMessageMaxLifetimeMs` | number | 604800000 | Max lifetime while waiting for MLS session establishment (7 days) | +| `edgeDrivenUnreachableDm` | boolean | false | Opt-in flood control for durably-unreachable DMs (see below) | + +**`edgeDrivenUnreachableDm` (default `false`)** changes how a direct message to +a peer that stays unreachable is retried. With the default, the SDK honors the +["a parked message never goes fully quiet"](message-delivery.md#reachability-probing) +contract: it keeps a timed reachability probe running on every carrier forever +(escalating to one per 600s). Set it to `true` and, after the probe has run a +bounded number of times, the message stops being timed-probed and rests in the +outbox — re-driven only when the peer next proves reachable (an inbound frame or +a presence-online edge flushes it), and a restart skips re-driving a +durably-failing backlog. Enabling the flag also bounds the core resend rate to +gone peers, which is what prevents a large unreachable backlog from tripping a +relay's rate limiter into a disconnect loop. + +This trades the SDK's "self-recovers even for a silent returning peer that never +advertises presence" guarantee for zero steady-state relay traffic to gone +peers. Enable it only for deployments whose peers always interact or advertise +presence when they return (for example a machine-to-machine capability +exchange). **Do not enable it for consumers that rely on the timed probe as +their only recovery path** — a pure-relay peer that comes back silently and +never polls presence would not be re-driven. The flag is off by default so every +existing native and third-party integration is unaffected. **Fixed (not configurable) message-plane limits**, listed here because they can surface as errors or as `message_failed` events: diff --git a/docs/message-delivery.md b/docs/message-delivery.md index 0694f4a6..7765b8dd 100644 --- a/docs/message-delivery.md +++ b/docs/message-delivery.md @@ -230,14 +230,18 @@ A parked message is re-driven with a fresh ACK budget on every reachability edge - The peer being discovered on a local transport - The peer coming online per presence — `internet_presence_watchlist()` includes recipients of pending/parked outbox messages, so the SDK owns presence-watching its own outbox; apps do not need their own watch queue for offline sends -**Reachability probing**: a parked message never goes fully quiet — the SDK keeps a timed reachability probe running on **every** carrier. The probe interval escalates with each consecutive unreachable park (15s doubling up to a 600s cap) and resets on any reachability edge. The escalation counter is per **recipient** while the probes are per message: a burst of DMs to one offline peer climbs the shared ladder once per park, so later messages start at an already-escalated interval rather than each walking 15s → 600s on their own — the delivery re-drive below is the compensating edge. If a probe attempt exhausts its ACK budget while the recipient still holds a live park counter, the message re-parks at the escalated interval rather than settling. +### Reachability probing + +A parked message never goes fully quiet — the SDK keeps a timed reachability probe running on **every** carrier. The probe interval escalates with each consecutive unreachable park (15s doubling up to a 600s cap) and resets on any reachability edge. The escalation counter is per **recipient** while the probes are per message: a burst of DMs to one offline peer climbs the shared ladder once per park, so later messages start at an already-escalated interval rather than each walking 15s → 600s on their own — the delivery re-drive below is the compensating edge. If a probe attempt exhausts its ACK budget while the recipient still holds a live park counter, the message re-parks at the escalated interval rather than settling. The probe is deliberately carrier-agnostic. With a local mesh carrier (BLE / Wi-Fi Direct) up the peer may be a room away even though the relay reports it offline — and possibly already a discovered neighbor, so no future edge would fire for it. On an internet-only device the external edges above are the *only* other recovery, which leaves delivery hostage to the platform's presence-polling cadence (and to nothing at all for a consumer that never polls presence). Probing over the relay is self-limiting in every outcome: a still-offline peer returns a fresh verdict that escalates the interval, an accepted frame becomes an ordinary in-flight send on the ACK ladder, and a peer that is back means the probe *was* the delivery. -Relay traffic is bounded differently in each of those branches. When the relay answers with a verdict, the escalation is the bound — one frame per interval per parked message, settling at one per 600s. When the relay *accepts* the frame instead (its push fallback succeeded, so no verdict comes back), the probe rides the ordinary ACK ladder — up to `max_retries` sends on 1s → 300s backoff, roughly 800s cumulative on the defaults — before re-parking at the escalated interval. Plan relay capacity against the second number, not the first. Delivery of any one message to a parked peer immediately re-drives that peer's remaining parked messages rather than leaving them on their own escalated timers. +Relay traffic is bounded differently in each of those branches. When the relay answers with a verdict, the escalation is the bound — one frame per interval per parked message, settling at one per 600s. When the relay *accepts* the frame instead (its push fallback succeeded, so no verdict comes back), the probe rides the ordinary ACK ladder — up to `max_retries` sends on 1s → 300s backoff, roughly 800s cumulative on the defaults — before re-parking at the escalated interval. That ~800s is the *fresh-entry* bound: a resend that has to register a new ACK carries the retry-queue entry's accumulated retry count onto it (`retry_count + 1`) rather than restarting the ladder at 0, so a probe that already climbed the backoff resumes near its current position and re-parks after roughly one more timeout — the carried case only sends *less*. Plan relay capacity against the fresh-entry number, not the verdict one. Without this carry-forward a never-ACKing (offline) recipient's probe would pin the resend delay at its 1s floor and flood the relay once per second. Delivery of any one message to a parked peer immediately re-drives that peer's remaining parked messages rather than leaving them on their own escalated timers. The outbox lifetime bounds the entry itself, with one caveat worth knowing: each probe refreshes the entry's last-send timestamp, so the sliding 7-day window stops binding and terminal `message_failed` moves out to the absolute cap (4× the lifetime, i.e. ~28 days). +**Opt-out for deployments where returning peers always interact.** The perpetual probe above is the default and stays the default. A deployment whose peers always send an inbound frame or advertise presence when they return (for example a machine-to-machine capability exchange) can set [`edgeDrivenUnreachableDm`](configuration.md#reliability-configuration) to `true`. The message is then timed-probed only a bounded number of times before it goes quiet and rests in the outbox, re-driven purely on the reachability edges above, and enabling the flag also caps the core resend rate to gone peers so a large unreachable backlog cannot trip a relay's rate limiter into a disconnect loop. This suspends the "never goes fully quiet" guarantee for unreachable DMs, so it must not be enabled for a consumer whose only recovery path is the timed probe (a silent returning peer that never polls presence would not be re-driven). It is off by default; the always-on behavior in this document is what every unconfigured integration sees. + **Raising `outbox_max_lifetime_ms` interacts with control-frame freshness.** A retransmitted control frame carries the signature it was minted with, timestamp included, and a receiver refuses one stamped more than 30 days ago (see [Control messages](spec/control-messages.md#freshness)). That window was chosen to clear the ~28-day absolute cap above, so the defaults are safe with room to spare. Configure a lifetime past a quarter of the window (7.5 days) and this device's own late retransmissions of `__CONN_REQ__` and the other signed control frames start being refused as stale by the peer they finally reach. Ordinary messages are unaffected: `__MLS_ENC__` is data-plane and carries no such signature. If a deployment genuinely needs a longer outbox, raise it for the reachability behaviour and expect control frames near the far end of the ladder to be dropped rather than delivered. **What parks and what doesn't**: