feat(client): report a 429 stream error as a StreamError event

src/client/node_io.rs

@@ -1905,6 +1905,18 @@ impl Client {
                     // Deliberate rate-limit backoff: the stability reset must
                     // not erase it even if the connection had been up >= 30s.
                     self.backoff_reset_suppressed.store(true, Ordering::Relaxed);
+                    // Not fidelity: WA Web (Handle/StreamError.js) special-cases
+                    // only 500..600, so 429 is indistinguishable from any other
+                    // reconnect there — survivable because a human watches the UI.
+                    // An embedder has none, so report the rate limit through
+                    // `StreamError`. Dispatched after the stores so a handler
+                    // sees the rate-limited session.
+                    self.core.event_bus.dispatch(Event::StreamError(
+                        crate::types::events::StreamError::builder()
+                            .code(code.to_string())
+                            .raw(node.to_owned())
+                            .build(),
+                    ));
                 }
                 "503" => {
                     // Server is going down/restarting: mark logged-out so sends fail

src/client/tests.rs

@@ -3260,6 +3260,95 @@ async fn test_stream_error_429_keeps_reconnect_with_backoff() {
     );
 }
 
+/// A rate-limited session parks the client for minutes; without an event the
+/// only trace is the missing connection. WA Web has no 429 arm to copy here
+/// (only 500..600 is special-cased), so this is our own `StreamError` contract
+/// applied consistently, and the neighbours must keep their own events.
+#[tokio::test]
+async fn test_stream_error_429_dispatches_stream_error_event() {
+    use wacore::types::events::{Event, EventHandler};
+
+    let client = create_offline_sync_test_client().await;
+    client.is_logged_in.store(true, Ordering::Relaxed);
+    let collector = Arc::new(crate::test_utils::TestEventCollector::default());
+    client
+        .subscribe_handler(collector.clone() as Arc<dyn EventHandler>)
+        .detach();
+
+    let node = NodeBuilder::new("stream:error")
+        .attr("code", "429")
+        .children([NodeBuilder::new("text")
+            .attr("text", "rate-overlimit")
+            .build()])
+        .build();
+    client.handle_stream_error(&node.as_node_ref()).await;
+
+    let events = collector.events();
+    let stream_errors: Vec<_> = events
+        .iter()
+        .filter_map(|event| match &**event {
+            Event::StreamError(stream_error) => Some(stream_error),
+            _ => None,
+        })
+        .collect();
+    assert_eq!(
+        stream_errors.len(),
+        1,
+        "429 must dispatch exactly one StreamError, got {events:?}"
+    );
+    assert_eq!(stream_errors[0].code, "429");
+    let raw = stream_errors[0]
+        .raw
+        .as_ref()
+        .expect("the stanza must ride along so the reason survives");
+    assert_eq!(raw.tag, "stream:error");
+    assert!(
+        raw.get_optional_child("text").is_some(),
+        "the raw stanza must keep the server's children, not just the code"
+    );
+    assert!(
+        !events
+            .iter()
+            .any(|event| matches!(**event, Event::LoggedOut(_) | Event::StreamReplaced(_))),
+        "429 is not a logout or a replacement"
+    );
+}
+
+/// The branches either side of 429 keep dispatching what they always did — a
+/// regression here would look like the 429 event working while 516/409 lost
+/// theirs.
+#[tokio::test]
+async fn test_stream_error_neighbours_keep_their_events() {
+    use wacore::types::events::{Event, EventHandler};
+
+    for (code, expect_logged_out) in [("516", true), ("401", true), ("409", false)] {
+        let client = create_offline_sync_test_client().await;
+        let collector = Arc::new(crate::test_utils::TestEventCollector::default());
+        client
+            .subscribe_handler(collector.clone() as Arc<dyn EventHandler>)
+            .detach();
+
+        let node = NodeBuilder::new("stream:error").attr("code", code).build();
+        client.handle_stream_error(&node.as_node_ref()).await;
+
+        let events = collector.events();
+        let matched = events.iter().any(|event| {
+            if expect_logged_out {
+                matches!(**event, Event::LoggedOut(_))
+            } else {
+                matches!(**event, Event::StreamReplaced(_))
+            }
+        });
+        assert!(matched, "{code} lost its event, got {events:?}");
+        assert!(
+            !events
+                .iter()
+                .any(|event| matches!(**event, Event::StreamError(_))),
+            "{code} must not also report as a generic StreamError"
+        );
+    }
+}
+
 #[tokio::test]
 async fn test_stream_error_503_keeps_reconnect() {
     let client = create_offline_sync_test_client().await;

src/prekeys.rs

@@ -724,6 +724,13 @@ impl Client {
     /// Retry schedule: 1s, 2s, 3s, 5s, 8s, 13s, ... capped at 610s.
     /// Verified against WA Web JS: `{ algo: { type: "fibonacci", first: 1e3, second: 2e3 }, max: 61e4 }`
     ///
+    /// `max` caps the delay, not the attempt count: `WAWebUploadPreKeysJob` ends its
+    /// loop only by calling `endWithValue` on `success`, and its error arms (>=500,
+    /// 406, anything else — 429 lands in the last) all fall through to the same
+    /// retry. So the absence of an attempt limit here is the mirror, not a gap; the
+    /// disconnect bail below is an exit WA Web does not even have (it awaits
+    /// `waitForConnection()` instead).
+    ///
     /// When `force` is true, bypasses the count guard (used by digest repair path).
     pub(crate) async fn upload_pre_keys_with_retry(
         &self,
@@ -776,7 +783,10 @@ impl Client {
                         ));
                     }
 
-                    let next = delay_a + delay_b;
+                    // Clamped like `fibonacci_backoff`: the sleep is already
+                    // capped, so past MAX the state only exists to overflow
+                    // (u64 at ~90 retries, a debug panic).
+                    let next = delay_a.saturating_add(delay_b).min(MAX_DELAY_SECS);
                     delay_a = delay_b;
                     delay_b = next;
                 }