Fix NATS classifier publish dropping messages with quoted multi-word args

A shlex-quoted plugin argument (e.g. choose 'multi word option') produces
a routing key containing a literal space. AMQP's topic exchange tolerates
that, but NATS's whitespace-delimited wire protocol does not - the PUB
frame gets corrupted and the server silently drops the connection before
the message reaches any subscriber. Sanitize whitespace out of the
subject before publishing; the JSON payload plugins actually read from is
untouched.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

diff --git a/src/distbot/common/broker_nats.py b/src/distbot/common/broker_nats.py
index 824e71de923b7b8b734e6cc25c9e6726c50594e9..cf37b67235cbd85a21d2713d1b4657609085e212 100644 (file)
--- a/src/distbot/common/broker_nats.py
+++ b/src/distbot/common/broker_nats.py
@@ -25,6 +25,7 @@ Design notes (see the migration plan for the full rationale):
 """
 import asyncio
 import logging
 """
 import asyncio
 import logging

+import re

 import threading
 from typing import Callable, List
 
 import threading
 from typing import Callable, List
 


@@ -48,6 +49,22 @@ def _to_bytes(value) -> bytes:
     return value.encode("utf-8") if isinstance(value, str) else value
 
 
     return value.encode("utf-8") if isinstance(value, str) else value
 
 

+_WHITESPACE = re.compile(r"\s+")
+
+
+def _sanitize_subject(subject: str) -> str:
+    # Routing keys come from shlex-quoted user text (e.g. "choose 'multi
+    # word option'") and can legally contain spaces under AMQP's topic
+    # exchange, which treats a routing key as an arbitrary string. NATS's
+    # wire protocol is whitespace-delimited ("PUB <subject> <#bytes>"), so a
+    # literal space inside a subject corrupts the PUB frame and the server
+    # drops the connection - confirmed live, message never reaches any
+    # subscriber. Since '#' is translated to ".*" by routing.matches() and
+    # absorbs any character here, substituting whitespace doesn't affect
+    # dispatch matching.
+    return _WHITESPACE.sub("_", subject)
+
+

 def _stream_name_for(queue: str) -> str:
     return queue.upper()
 
 def _stream_name_for(queue: str) -> str:
     return queue.upper()
 


@@ -152,7 +169,7 @@ class NatsSyncBroker(SyncBroker):
             self._loop.close()
 
     def publish(self, subject: str, body: bytes) -> None:
             self._loop.close()
 
     def publish(self, subject: str, body: bytes) -> None:

-        full_subject = CLASSIFIER_PREFIX + _to_str(subject)
+        full_subject = CLASSIFIER_PREFIX + _sanitize_subject(_to_str(subject))

         self._call(self._nc.publish(full_subject, _to_bytes(body)))
 
     def publish_queue(self, queue: str, body: bytes, durable: bool = True, max_age_seconds: int = None) -> None:
         self._call(self._nc.publish(full_subject, _to_bytes(body)))
 
     def publish_queue(self, queue: str, body: bytes, durable: bool = True, max_age_seconds: int = None) -> None:

diff --git a/tests/test_unit/test_broker_nats.py b/tests/test_unit/test_broker_nats.py
new file mode 100644 (file)
index 0000000..18cd7ab
--- /dev/null
+++ b/tests/test_unit/test_broker_nats.py
@@ -0,0 +1,20 @@
+# -*- coding: utf-8 -*-
+import pytest
+
+from distbot.common.broker_nats import _sanitize_subject
+
+
+@pytest.mark.parametrize("subject,expected", [
+    # A quoted, multi-word plugin argument (e.g. choose "'multi word
+    # option'") shows up as a single shlex token containing spaces, which
+    # AMQP's topic exchange tolerates as an arbitrary routing key but NATS's
+    # whitespace-delimited wire protocol ("PUB <subject> <#bytes>") does
+    # not - an unsanitized space here corrupts the PUB frame and the
+    # message never reaches any subscriber (confirmed against a live
+    # nats-server).
+    ("nick.choose.will ich nicht", "nick.choose.will_ich_nicht"),
+    ("nick.choose.a\tb", "nick.choose.a_b"),
+    ("nick.dice.5", "nick.dice.5"),
+])
+def test_sanitize_subject_strips_whitespace(subject, expected):
+    assert _sanitize_subject(subject) == expected