feat(transports): give inbound message pacing a voice, so an opt-in ceiling can be tuned (BACKLOG #290) (#1229)

Message-rate pacing ships OFF by the 2026-08-11 ruling, on the ground that a safe
number can only come from a site's own feed profile. That posture needs the
operator to be able to watch a number engage, and pacing was silent by
construction: it never drops, NAKs, refuses or errors, so a paced interface was
indistinguishable from a slow one and nothing was written anywhere.

_MessagePacer now tallies each APPLIED read delay and reports it at WARNING, at
most once per 60s. Instrumented at the two places a wait is acted on -- pace()
for the stream pair and deficit() for the listener pair -- never in charge(),
which both route through and which deficit() consults on every read; counting
there would tally one outstanding debt once per consult.

Carries the connection name through from Source.name so a report can be traced
to a feed. Metadata only: name, count, duration, configured rate.

No default changes and nothing paces differently.

Co-authored-by: wshallwshall <mefordev@messagefoundry.org>

Author: wshallwshall

messagefoundry/transports/http_listener.py

@@ -470,7 +470,9 @@ def __init__(self, config: Source) -> None:
         #: pace()/settle() for the same reason — there is no next read on this connection to settle
         #: against. Distinct from the injected `intake_rate_limiter`, which REFUSES failed auth
         #: attempts with a 429; this one only ever waits.
-        self._pacer = _MessagePacer.for_rate(self.max_messages_per_second, self.message_burst)
+        self._pacer = _MessagePacer.for_rate(
+            self.max_messages_per_second, self.message_burst, name=config.name or ""
+        )
         # Per-connection peer-IP allowlist (Tier 4): refuse a non-listed peer at accept (fail-closed).
         # Absent/empty = no restriction. Mirrors MLLPSource.
         sa = s.get("source_ip_allowlist")

messagefoundry/transports/mllp.py

@@ -116,6 +116,12 @@
 #: mechanism exists and an operator opts in with their own number. The cell stays `partial` on the
 #: shipped default and the record says why; that is the honest outcome, not a disappointing one.
 DEFAULT_MAX_MESSAGES_PER_SECOND: float | None = None
+
+#: Seconds between operator-facing pacing reports on ONE pacer (BACKLOG #290). Pacing is silent by
+#: construction — it never drops, NAKs or errors — so without a report an operator cannot tell a
+#: paced interface from a slow one. A pacer in deficit is consulted on every read, so the report is
+#: throttled to this window; see :meth:`_MessagePacer._note_paced`.
+_PACING_REPORT_SECONDS = 60.0
 # On stop()/reload, established clients are closed and their handlers given this long to finish an
 # in-flight commit before the connection tasks are cancelled — bounds shutdown so a peer holding a
 # connection open can't hang it (review H-2).
@@ -1395,16 +1401,36 @@ class _MessagePacer:
     and drives it with :meth:`deficit` / :meth:`charge`. There is no flag and no branch in here.
     """
 
-    __slots__ = ("_capacity", "_last", "_pending_wait", "_rate", "_tokens")
-
-    def __init__(self, rate: float, burst: float, *, now: float) -> None:
+    __slots__ = (
+        "_capacity",
+        "_last",
+        "_name",
+        "_paced_count",
+        "_paced_seconds",
+        "_pending_wait",
+        "_rate",
+        "_report_at",
+        "_tokens",
+    )
+
+    def __init__(self, rate: float, burst: float, *, now: float, name: str = "") -> None:
         self._rate = rate
         self._capacity = max(burst, 1.0)
         self._tokens = self._capacity
         self._last = now
         #: Debt owed before the next read, in seconds. PRIVATE, and settled only through pace() —
         #: a consult-then-clear a caller performs by hand is an invariant restated once per loop.
         self._pending_wait = 0.0
+        #: The declaring inbound's name, carried only so a pacing report can NAME the connection an
+        #: operator has to go and look at (the :attr:`Source.name` precedent). "" when unwired.
+        self._name = name
+        #: Applied-delay tally since the last report, reset by each report.
+        self._paced_count = 0
+        self._paced_seconds = 0.0
+        #: Next monotonic stamp a report may be emitted at. Starts at ``now`` so the FIRST time
+        #: pacing engages is reported immediately — that transition is the event an operator most
+        #: needs, and holding it back for a window would hide it behind the throttle.
+        self._report_at = now
 
     def charge(self, messages: int, *, now: float) -> float:
         """Charge ``messages`` and return the seconds to wait before reading again (0.0 if none).
@@ -1428,6 +1454,7 @@ async def pace(self) -> None:
         """
         if (wait := self._pending_wait) > 0.0:
             self._pending_wait = 0.0
+            self._note_paced(wait)
             await asyncio.sleep(wait)
 
     def settle(self, messages: int) -> None:
@@ -1446,16 +1473,61 @@ def deficit(self, *, now: float) -> float:
         connections has to recompute it at read time instead. Delegates to :meth:`charge` rather
         than re-deriving the arithmetic, so the two can never disagree.
         """
-        return self.charge(0, now=now)
+        owed = self.charge(0, now=now)
+        if owed > 0.0:
+            self._note_paced(owed, now=now)
+        return owed
+
+    def _note_paced(self, seconds: float, *, now: float | None = None) -> None:
+        """Tally one APPLIED read delay and report it to the operator, throttled.
+
+        Called from the two places a wait is actually acted on — :meth:`pace` for the stream pair and
+        :meth:`deficit` for the listener pair — never from :meth:`charge`, which both of those route
+        through and which ``deficit`` consults on every read. Counting in ``charge`` would tally the
+        same outstanding debt once per consult and report a number that is not a count of anything.
+
+        **Why pacing needs a voice at all.** The control is otherwise entirely invisible: it never
+        drops, NAKs, refuses or errors, so a paced interface looks to the operator exactly like a
+        slow one, and nothing is written anywhere. The 2026-08-11 ruling ships
+        :data:`DEFAULT_MAX_MESSAGES_PER_SECOND` OFF *because* a safe number can only come from a
+        site's own feed profile — and a site cannot tune a number it has no way to watch engage.
+        Reporting is the half that makes the opt-in posture usable; it changes no default and paces
+        nothing differently.
+
+        Throttled to one line per :data:`_PACING_REPORT_SECONDS` because a pacer that is in deficit
+        is consulted on every read, and an unthrottled line would restate one fact thousands of
+        times. WARNING rather than INFO: a clinical interface being held back is an operator-facing
+        condition, not routine chatter.
+
+        **Metadata only.** The connection name, a count, a duration and the configured rate — never a
+        frame, a peer address, or a byte of the body being paced (PHI.md; CLAUDE.md section 9).
+        """
+        self._paced_count += 1
+        self._paced_seconds += seconds
+        stamp = time.monotonic() if now is None else now
+        if stamp < self._report_at:
+            return
+        logger.warning(
+            "inbound message pacing engaged on %s: %d read delay(s) totalling %.3fs "
+            "(max_messages_per_second=%g). The sender is being held back, not refused — no message "
+            "is dropped. Raise the rate if this feed is legitimate.",
+            self._name or "<unnamed inbound>",
+            self._paced_count,
+            self._paced_seconds,
+            self._rate,
+        )
+        self._paced_count = 0
+        self._paced_seconds = 0.0
+        self._report_at = stamp + _PACING_REPORT_SECONDS
 
     @classmethod
-    def for_rate(cls, rate: float | None, burst: float) -> _MessagePacer | None:
+    def for_rate(cls, rate: float | None, burst: float, *, name: str = "") -> _MessagePacer | None:
         """Build a pacer, or ``None`` when no rate is configured — the shipped default.
 
         The single place the off-default is honoured, so the four intakes that pace cannot drift
         apart on what "unset" means. See :data:`DEFAULT_MAX_MESSAGES_PER_SECOND` for why off.
         """
-        return cls(rate, burst, now=time.monotonic()) if rate else None
+        return cls(rate, burst, now=time.monotonic(), name=name) if rate else None
 
 
 def _pacing_settings(settings: Mapping[str, Any]) -> tuple[float | None, float]:
@@ -1494,6 +1566,8 @@ def __init__(self, config: Source) -> None:
         # Message-rate pacing. Absent -> OFF, unlike the caps above; see _pacing_settings and
         # DEFAULT_MAX_MESSAGES_PER_SECOND for why that deviation is deliberate and ruled.
         self.max_messages_per_second, self.message_burst = _pacing_settings(s)
+        # Carried only so a pacing report can name this connection (BACKLOG #290).
+        self._pacing_name = config.name or ""
         # Per-connection peer-IP allowlist (Tier 4 operability): when set, a connecting peer whose IP
         # is not listed is refused at accept time. Absent/empty = no restriction.
         sa = s.get("source_ip_allowlist")
@@ -1629,7 +1703,9 @@ async def _on_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamW
             await self._emit_event("established", peer_host=peer_host)
             try:
                 decoder = MLLPDecoder(max_frame_bytes=self.max_frame_bytes)
-                pacer = _MessagePacer.for_rate(self.max_messages_per_second, self.message_burst)
+                pacer = _MessagePacer.for_rate(
+                    self.max_messages_per_second, self.message_burst, name=self._pacing_name
+                )
                 while True:
                     # ASVS 2.4.1 / 15.2.2 — the wait is BEFORE the read, never around the handler.
                     if pacer is not None:

messagefoundry/transports/tcp.py

@@ -450,6 +450,8 @@ def __init__(self, config: Source) -> None:
         # port changed REACHABILITY (raw TCP had no rate control in any configuration), never the
         # default -- a stock raw-TCP inbound still has no rate bound.
         self.max_messages_per_second, self.message_burst = _pacing_settings(s)
+        # Carried only so a pacing report can name this connection (BACKLOG #290).
+        self._pacing_name = config.name or ""
         # Per-connection peer-IP allowlist (Tier 4 operability): refuse a non-listed peer at accept.
         # Absent/empty = no restriction. Mirrors MLLPSource.
         sa = s.get("source_ip_allowlist")
@@ -568,7 +570,9 @@ async def _on_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamW
             await self._emit_event("established", peer_host=peer_host)
             try:
                 decoder = self.codec.decoder(max_frame_bytes=self.max_frame_bytes)
-                pacer = _MessagePacer.for_rate(self.max_messages_per_second, self.message_burst)
+                pacer = _MessagePacer.for_rate(
+                    self.max_messages_per_second, self.message_burst, name=self._pacing_name
+                )
                 while True:
                     # ASVS 2.4.1 / 15.2.2 — the wait is BEFORE the read, never around the handler.
                     if pacer is not None:

messagefoundry/transports/x12.py

@@ -501,6 +501,8 @@ def __init__(self, config: Source) -> None:
         # token per INTERCHANGE, which is what this connector's frame is. The port changed
         # REACHABILITY, never the default -- a stock X12 inbound still has no rate bound.
         self.max_messages_per_second, self.message_burst = _pacing_settings(s)
+        # Carried only so a pacing report can name this connection (BACKLOG #290).
+        self._pacing_name = config.name or ""
         # Per-connection peer-IP allowlist (Tier 4 operability): refuse a non-listed peer at accept.
         # Absent/empty = no restriction. Mirrors TcpSource/MLLPSource.
         sa = s.get("source_ip_allowlist")
@@ -592,7 +594,9 @@ async def _on_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamW
             self._active += 1
             try:
                 decoder = X12FrameReader(max_interchange_bytes=self.max_interchange_bytes)
-                pacer = _MessagePacer.for_rate(self.max_messages_per_second, self.message_burst)
+                pacer = _MessagePacer.for_rate(
+                    self.max_messages_per_second, self.message_burst, name=self._pacing_name
+                )
                 while True:
                     # ASVS 2.4.1 / 15.2.2 — the wait is BEFORE the read, never around the handler.
                     if pacer is not None:

tests/test_mllp_message_pacing.py

@@ -19,6 +19,8 @@
 from __future__ import annotations
 
 import asyncio
+import logging
+import time
 
 import pytest
 from _ingress_pace_probe import install_ingress_pace_probe, stream_debt_seconds
@@ -239,3 +241,88 @@ def test_the_shipped_default_constant_is_still_off() -> None:
     """`DEFAULT_MAX_MESSAGES_PER_SECOND` is what the connector falls back to when the key is absent.
     If it ever becomes non-None, exposing the keys would have silently turned pacing on for everyone."""
     assert DEFAULT_MAX_MESSAGES_PER_SECOND is None
+
+
+# --- BACKLOG #290: pacing has to be observable, or its opt-in posture cannot be tuned -------------
+#
+# The 2026-08-11 ruling ships DEFAULT_MAX_MESSAGES_PER_SECOND OFF because a safe number can only come
+# from a site's own feed profile. That posture only works if an operator who sets a number can watch
+# it engage -- and pacing is silent by construction (it never drops, NAKs, refuses or errors), so a
+# paced interface is indistinguishable from a slow one. These pin the report, not the pacing.
+
+
+def test_pacing_reports_itself_on_a_stream_intake(caplog: pytest.LogCaptureFixture) -> None:
+    """The pace()/settle() pair -- MLLP, raw TCP and X12 -- reports the delay it applies.
+
+    Fails without the change: `_note_paced` does not exist, so `pace()` sleeps in silence and nothing
+    reaches the log at any level.
+    """
+    pacer = _MessagePacer(1000.0, 1.0, now=time.monotonic(), name="IB_ACME_ADT")
+    pacer.settle(5)  # 5 messages against a burst of 1 -> 4 tokens of debt, 4ms at 1000/s
+    with caplog.at_level(logging.WARNING, logger="messagefoundry.transports.mllp"):
+        asyncio.run(pacer.pace())
+    assert "IB_ACME_ADT" in caplog.text
+    assert "pacing engaged" in caplog.text
+    # The operator has to be told this is a hold, not a loss -- the whole point of the control.
+    assert "not refused" in caplog.text
+
+
+def test_pacing_reports_itself_on_a_listener_scoped_intake(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """The deficit()/charge() pair -- the HTTP intake's one listener-wide bucket -- reports too.
+
+    Driven on a fully synthetic clock, so it pins the report rather than any wall-clock timing.
+    """
+    pacer = _MessagePacer(1.0, 1.0, now=0.0, name="IB_ACME_HTTP")
+    pacer.charge(5, now=0.0)  # 5 messages against a burst of 1 -> 4s of debt at 1/s
+    with caplog.at_level(logging.WARNING, logger="messagefoundry.transports.mllp"):
+        owed = pacer.deficit(now=0.0)
+    assert owed > 0.0
+    assert "IB_ACME_HTTP" in caplog.text
+
+
+def test_a_pacer_that_is_not_engaging_says_nothing(caplog: pytest.LogCaptureFixture) -> None:
+    """NEGATIVE CONTROL for the two tests above. Without this, a report emitted unconditionally --
+    on every read of every paced connection, whether or not the bucket is in deficit -- would pass
+    both of them while flooding the log of a connection that is under its rate and fine."""
+    pacer = _MessagePacer(1000.0, 1000.0, now=0.0, name="IB_QUIET")
+    with caplog.at_level(logging.WARNING, logger="messagefoundry.transports.mllp"):
+        assert pacer.deficit(now=0.0) == 0.0  # well inside the burst -> nothing owed
+        asyncio.run(pacer.pace())  # no debt held -> no sleep, no report
+    assert caplog.text == ""
+
+
+def test_the_report_is_throttled_to_one_line_per_window(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A pacer in deficit is consulted on EVERY read, so an unthrottled line would restate one fact
+    thousands of times and bury the log. Two applied delays inside one window produce one line."""
+    pacer = _MessagePacer(1.0, 1.0, now=0.0, name="IB_BUSY")
+    pacer.charge(5, now=0.0)
+    with caplog.at_level(logging.WARNING, logger="messagefoundry.transports.mllp"):
+        first = pacer.deficit(now=0.0)  # stamp == _report_at -> reports
+        second = pacer.deficit(now=1.0)  # inside the window -> tallied, silent
+    assert first > 0.0 and second > 0.0
+    assert len([r for r in caplog.records if "pacing engaged" in r.getMessage()]) == 1
+
+
+def test_the_report_names_the_connection_the_source_was_wired_with() -> None:
+    """The name is not decoration: a pacing report an operator cannot trace to a connection tells
+    them a feed somewhere is being held back and nothing about which one. Pins the wiring from
+    `Source.name` through the source to the pacer, which is the half a pacer-only test cannot see."""
+    src = _source(max_messages_per_second=5.0)
+    assert src._pacing_name == "IB_TEST"
+    pacer = _MessagePacer.for_rate(
+        src.max_messages_per_second, src.message_burst, name=src._pacing_name
+    )
+    assert pacer is not None
+    assert pacer._name == "IB_TEST"
+
+
+def test_for_rate_still_returns_none_when_pacing_is_off() -> None:
+    """POSITIVE CONTROL on the test above. If `for_rate` had stopped honouring the off-default while
+    gaining its `name` argument, every pacing test in this file would still pass and pacing would
+    have been silently turned on for every inbound."""
+    assert _MessagePacer.for_rate(None, 1.0, name="IB_TEST") is None
+    assert _MessagePacer.for_rate(0, 1.0, name="IB_TEST") is None