Signal live adapter silent delivery failures and send acknowledgement validation

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #49312 fix(signal): reject silent per-recipient delivery failures in live adapter (#49260)

  merged to main

  files: cron/scheduler.py, gateway/platforms/signal.py, scripts/release.py, tests/gateway/test_signal.py

  Show the diff

  • cron/scheduler.py

    @@ -810,6 +810,7 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
             # rooms (e.g. Matrix) where the standalone HTTP path cannot encrypt.
             runtime_adapter = (adapters or {}).get(platform)
             delivered = False
    +        target_errors = []
             if runtime_adapter is not None and loop is not None and getattr(loop, "is_running", lambda: False)():
                 send_metadata = {"thread_id": thread_id} if thread_id else None
                 try:
    @@ -824,18 +825,26 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
                         )
                         if future is None:
                             adapter_ok = False
    +                        target_errors.append("live adapter event loop scheduling failed")
                         else:
                             try:
                                 send_result = future.result(timeout=60)
    -                        except TimeoutError:
    +                        except TimeoutError as te:
                                 future.cancel()
    +                            target_errors.append(f"live adapter send timed out: {te}")
                                 raise
    -                        if send_result and not getattr(send_result, "success", True):
    -                            err = getattr(send_result, "error", "unknown")
    +                        except Exception as ex:
    +                            target_errors.append(f"live adapter send failed: {ex}")
    +                            raise
    +
    +                        if send_result is None or not getattr(send_result, "success", True):
    +                            err = getattr(send_result, "error", "unknown") if send_result else "no response from adapter"
    +                            msg = f"live adapter send to {platform_name}:{chat_id} failed: {err}"
                                 logger.warning(
    -                                "Job '%s': live adapter send to %s:%s failed (%s), falling back to standalone",
    -                                job["id"], platform_name, chat_id, err,
    +                                "Job '%s': %s, falling back to standalone",
    +                                job["id"], msg,
                                 )
    +                            target_errors.append(msg)
                                 adapter_ok = False  # fall through to standalone path
                             elif (
                                 send_result
    @@ -867,9 +876,12 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
                         logger.info("Job '%s': delivered to %s:%s via live adapter", job["id"], platform_name, chat_id)
                         delivered = True
                 except Exception as e:
    +                err_msg = f"live adapter delivery to {platform_name}:{chat_id} failed: {e}"
    +                if not any(err_msg in err for err in target_errors):
    +                    target_errors.append(err_msg)
                     logger.warning(
    -                    "Job '%s': live adapter delivery to %s:%s failed (%s), falling back to standalone",
    -                    job["id"], platform_name, chat_id, e,
    +                    "Job '%s': %s, falling back to standalone",
    +                    job["id"], err_msg,
                     )
     
             if not delivered:
    @@ -889,13 +901,15 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
                 except Exception as e:
                     msg = f"delivery to {platform_name}:{chat_id} failed: {e}"
                     logger.error("Job '%s': %s", job["id"], msg)
    -                delivery_errors.append(msg)
    +                target_errors.extend([msg])
    +                delivery_errors.extend(target_errors)
                     continue
     
                 if result and result.get("error"):
                     msg = f"delivery error: {result['error']}"
                     logger.error("Job '%s': %s", job["id"], msg)
    -                delivery_errors.append(msg)
    +                target_errors.extend([msg])
    +                delivery_errors.extend(target_errors)
                     continue
     
                 logger.info("Job '%s': delivered to %s:%s", job["id"], platform_name, chat_id)

  • gateway/platforms/signal.py

    @@ -796,7 +796,16 @@ async def _rpc(
                         logger.debug("Signal RPC error (%s): %s", method, err)
                     return None
     
    -            return data.get("result")
    +            result = data.get("result")
    +            if isinstance(result, dict) and raise_on_rate_limit:
    +                results = result.get("results")
    +                if isinstance(results, list):
    +                    for r in results:
    +                        if isinstance(r, dict) and r.get("type") == "RATE_LIMIT_FAILURE":
    +                            retry_after = r.get("retryAfterSeconds")
    +                            raise SignalRateLimitError("Rate limit exceeded for recipient", retry_after=retry_after)
    +
    +            return result
     
             except SignalRateLimitError:
                 raise
    @@ -960,6 +969,29 @@ def format_message(self, content: str) -> str:
             # Our send() override bypasses this entirely.
             return content
     
    +    def _validate_send_result(self, result: Any) -> tuple[bool, Optional[str]]:
    +        """Validate signal-cli send response results.
    +
    +        Returns (success, error_message).
    +        """
    +        if not result or not isinstance(result, dict):
    +            return True, None
    +
    +        results = result.get("results")
    +        if isinstance(results, list):
    +            for r in results:
    +                if not isinstance(r, dict):
    +                    continue
    +                rtype = r.get("type")
    +                if rtype and rtype != "SUCCESS":
    +                    return False, str(rtype)
    +                if "success" in r and not r.get("success"):
    +                    fail = r.get("failure")
    +                    if fail:
    +                        return False, str(fail)
    +                    return False, "Recipient delivery failed"
    +        return True, None
    +
         # ------------------------------------------------------------------
         # Sending
         # ------------------------------------------------------------------
    @@ -992,9 +1024,13 @@ async def send(
             else:
                 params["recipient"] = [await self._resolve_recipient(chat_id)]
     
    +        logger.info("[Signal] Sending response (%d chars) to %s", len(plain_text), chat_id)
             result = await self._rpc("send", params)
     
             if result is not None:
    +            success, err_msg = self._validate_send_result(result)
    +            if not success:
    +                return SendResult(success=False, error=err_msg, raw_response=result)
                 self._track_sent_timestamp(result)
                 # Signal has no editable message identifier. Returning None keeps the
                 # stream consumer on the non-edit fallback path instead of pretending
    @@ -1171,14 +1207,33 @@ async def send_multiple_images(
                         )
                         _rpc_duration = time.monotonic() - _rpc_t0
                         if result is not None:
    -                        self._track_sent_timestamp(result)
    -                        await scheduler.report_rpc_duration(_rpc_duration, n)
    -                        logger.info(
    -                            "Signal batch %d/%d: %d attachments sent in %.1fs "
    -                            "(attempt %d/%d)",
    -                            idx + 1, len(att_batches), n, _rpc_duration,
    -                            attempt, SIGNAL_RATE_LIMIT_MAX_ATTEMPTS,
    -                        )
    +                        success, err_msg = self._validate_send_result(result)
    +                        if success:
    +                            self._track_sent_timestamp(result)
    +                            await scheduler.report_rpc_duration(_rpc_duration, n)
    +                            logger.info(
    +                                "Signal batch %d/%d: %d attachments sent in %.1fs "
    +                                "(attempt %d/%d)",
    +                                idx + 1, len(att_batches), n, _rpc_duration,
    +                                attempt, SIGNAL_RATE_LIMIT_MAX_ATTEMPTS,
    +                            )
    +                        else:
    +                            logger.error(
    +                                "Signal: RPC send failed for batch %d/%d (%d attachments, "
    +                                "attempt %d/%d, rpc_duration=%.1fs): %s",
    +                                idx + 1, len(att_batches), n,
    +                                attempt, SIGNAL_RATE_LIMIT_MAX_ATTEMPTS,
    +                                _rpc_duration, err_msg,
    +                            )
    +                            # Retry transient (non-rate-limit) failures once
    +                            if attempt < SIGNAL_RATE_LIMIT_MAX_ATTEMPTS:
    +                                backoff = 2.0 ** attempt
    +                                logger.info(
    +                                    "Signal: retrying batch %d/%d after %.1fs backoff",
    +                                    idx + 1, len(att_batches), backoff,
    +                                )
    +                                await asyncio.sleep(backoff)
    +                                continue
                         else:
                             # Assume the server didn't accept the batch, don't deduce tokens
                             logger.error(
    @@ -1277,6 +1332,9 @@ async def send_image(
     
             result = await self._rpc("send", params)
             if result is not None:
    +            success, err_msg = self._validate_send_result(result)
    +            if not success:
    +                return SendResult(success=False, error=err_msg, raw_response=result)
                 self._track_sent_timestamp(result)
                 return SendResult(success=True)
             return SendResult(success=False, error="RPC send with attachment failed")
    @@ -1316,6 +1374,9 @@ async def _send_attachment(
     
             result = await self._rpc("send", params)
             if result is not None:
    +            success, err_msg = self._validate_send_result(result)
    +            if not success:
    +                return SendResult(success=False, error=err_msg, raw_response=result)
                 self._track_sent_timestamp(result)
                 return SendResult(success=True)
             return SendResult(success=False, error=f"RPC send {media_label.lower()} failed")

  • scripts/release.py

    @@ -48,6 +48,7 @@
         "[redacted:email]": "salesondemandio",
         "[redacted:email]": "victor-kyriazakos",
         "[email protected]": "JoaoMarcos44",
    +    "[redacted:email]": "JoaoMarcos44",
         "[email protected]": "srojk34",
         "[email protected]": "sitkarev",
         "[redacted:email]": "omegazheng",

  • tests/gateway/test_signal.py

    @@ -1009,6 +1009,97 @@ async def test_send_returns_none_message_id_for_non_dict(self, monkeypatch):
             assert result.message_id is None
     
     
    +class TestSignalSendResultValidation:
    +    """Verify that send() validates recipient-level delivery results."""
    +
    +    @pytest.mark.asyncio
    +    async def test_send_success_when_results_has_success(self, monkeypatch):
    +        adapter = _make_signal_adapter(monkeypatch)
    +        mock_rpc, _ = _stub_rpc({
    +            "timestamp": 1712345678000,
    +            "results": [
    +                {
    +                    "recipientAddress": {"number": "+155****4567"},
    +                    "type": "SUCCESS"
    +                }
    +            ]
    +        })
    +        adapter._rpc = mock_rpc
    +        adapter._stop_typing_indicator = AsyncMock()
    +
    +        result = await adapter.send(chat_id="+155****4567", content="hello")
    +        assert result.success is True
    +
    +    @pytest.mark.asyncio
    +    async def test_send_failure_when_results_has_failure_type(self, monkeypatch):
    +        adapter = _make_signal_adapter(monkeypatch)
    +        mock_rpc, _ = _stub_rpc({
    +            "timestamp": 1712345678000,
    +            "results": [
    +                {
    +                    "recipientAddress": {"number": "+155****4567"},
    +                    "type": "UNREGISTERED_FAILURE"
    +                }
    +            ]
    +        })
    +        adapter._rpc = mock_rpc
    +        adapter._stop_typing_indicator = AsyncMock()
    +
    +        result = await adapter.send(chat_id="+155****4567", content="hello")
    +        assert result.success is False
    +        assert result.error == "UNREGISTERED_FAILURE"
    +
    +    @pytest.mark.asyncio
    +    async def test_send_failure_when_results_has_success_false(self, monkeypatch):
    +        adapter = _make_signal_adapter(monkeypatch)
    +        mock_rpc, _ = _stub_rpc({
    +            "timestamp": 1712345678000,
    +            "results": [
    +                {
    +                    "recipientAddress": {"number": "+155****4567"},
    +                    "success": False,
    +                    "failure": "Some connection error"
    +                }
    +            ]
    +        })
    +        adapter._rpc = mock_rpc
    +        adapter._stop_typing_indicator = AsyncMock()
    +
    +        result = await adapter.send(chat_id="+155****4567", content="hello")
    +        assert result.success is False
    +        assert result.error == "Some connection error"
    +
    +    @pytest.mark.asyncio
    +    async def test_rpc_raises_rate_limit_on_results_failure(self, monkeypatch):
    +        adapter = _make_signal_adapter(monkeypatch)
    +        mock_client = AsyncMock()
    +        mock_response = MagicMock()
    +        mock_response.status_code = 200
    +        mock_response.json.return_value = {
    +            "jsonrpc": "2.0",
    +            "result": {
    +                "timestamp": 1712345678000,
    +                "results": [
    +                    {
    +                        "recipientAddress": {"number": "+155****4567"},
    +                        "type": "RATE_LIMIT_FAILURE",
    +                        "retryAfterSeconds": 15
    +                    }
    +                ]
    +            },
    +            "id": "1"
    +        }
    +        mock_client.post = AsyncMock(return_value=mock_response)
    +        adapter.client = mock_client
    +
    +        from gateway.platforms.signal_rate_limit import SignalRateLimitError
    +        with pytest.raises(SignalRateLimitError) as exc_info:
    +            await adapter._rpc("send", {"recipient": ["+155****4567"]}, raise_on_rate_limit=True)
    +
    +        assert "Rate limit exceeded for recipient" in str(exc_info.value)
    +        assert exc_info.value.retry_after == 15
    +
    +
     # ---------------------------------------------------------------------------
     # stop_typing() delegates to _stop_typing_indicator (#4647)
     # ---------------------------------------------------------------------------