Fix Rate Drop behaviour over FBP/OBP.

bridge_master.py

@@ -117,6 +117,9 @@ TIMER_OPTION_KEYS = (
 )
 TIMER_EXPLICIT_KEY = '_TIMER_OPTIONS_SET'
 DMR_ID_MAX = 16777215
+OBP_RATE_DROP_MIN_DURATION = 2.0
+OBP_RATE_DROP_MIN_PACKETS = 50
+OBP_RATE_DROP_MAX_PPS = 50
 EMB_LC_TA_HEADER = 0x04
 EMB_LC_TA_BLOCKS = {0x05: 1, 0x06: 2, 0x07: 3}
 EMB_LC_GPS_INFO = 0x08
@@ -2120,9 +2123,11 @@ class routerOBP(OPENBRIDGE):
                 
                 #Rate drop
                 call_duration = pkt_time - self.STATUS[_stream_id]['START']
-                if call_duration and self.STATUS[_stream_id]['packets'] > 18 and (self.STATUS[_stream_id]['packets'] / call_duration > 25):
-                    logger.warning("(%s) *PacketControl* RATE DROP! Stream ID:, %s TGID: %s",self._system,int_id(_stream_id),int_id(_dst_id))
-                    self.proxy_BadPeer()
+                packet_rate = 0
+                if call_duration:
+                    packet_rate = self.STATUS[_stream_id]['packets'] / call_duration
+                if call_duration >= OBP_RATE_DROP_MIN_DURATION and self.STATUS[_stream_id]['packets'] > OBP_RATE_DROP_MIN_PACKETS and packet_rate > OBP_RATE_DROP_MAX_PPS:
+                    logger.warning("(%s) *PacketControl* RATE DROP! Stream ID: %s TGID: %s PEER: %s SRC: %s PACKETS: %s DURATION: %.2f RATE: %.2f/s",self._system,int_id(_stream_id),int_id(_dst_id),int_id(_peer_id),int_id(_source_server),self.STATUS[_stream_id]['packets'],call_duration,packet_rate)
                     return
                 
                 #Duplicate handling#

docs/test-harness-design.md

@@ -217,7 +217,10 @@ for rule timeout checks without sleeping.
   raise during local packet-rate calculation before duplicate/drop handling can
   run.
 - OBP group voice rate-control bugs: per-stream packet-rate protection should
-  use elapsed stream duration, not the absolute stream start timestamp.
+  use elapsed stream duration, not the absolute stream start timestamp. FBP/OBP
+  trunk-side rate control should tolerate short bursts and avoid HBP proxy
+  blacklisting behaviour while still dropping sustained abnormal per-stream
+  rates.
 - OBP voice lifecycle/report-coupling bugs: terminators should mark streams
   finished even when live reporting is disabled, so late same-stream packets do
   not route because a dashboard option is off.
@@ -431,8 +434,8 @@ test venv; otherwise a temporary venv is used for the scenario.
 - Transport serialization bugs after `bridge_master.py` calls `send_system()`.
 - Bugs caused by FreeDMR startup config, process lifecycle, Twisted reactor
   scheduling or socket binding.
-- Cadence-sensitive bugs: packet-rate limiting, duplicate/out-of-order handling
-  under realistic arrival spacing and jitter.
+- Cadence-sensitive bugs: packet-rate limiting, FBP trunk burst tolerance,
+  duplicate/out-of-order handling under realistic arrival spacing and jitter.
 - Regressions against FreeDMR's real-time discard model: delayed packets should
   not be re-emitted in corrected order or override loop-control/source-quench
   decisions.

docs/testing.md

@@ -144,9 +144,11 @@ GPS LC cycles are decoded through the standalone validated codec to system log
 messages without changing packet routing or mutation behaviour.
 OBP group voice rate-drop coverage verifies per-stream packet-rate protection is
 calculated from elapsed stream duration rather than the absolute stream start
-timestamp. OBP voice lifecycle coverage verifies a voice terminator marks the
-stream finished even when live reporting is disabled, so late packets with the
-same stream ID are suppressed independently of dashboard configuration.
+timestamp, ignores short FBP bursts, and only drops sustained over-rate FBP
+streams using a more lenient trunk-side threshold than HBP. OBP voice lifecycle
+coverage verifies a voice terminator marks the stream finished even when live
+reporting is disabled, so late packets with the same stream ID are suppressed
+independently of dashboard configuration.
 OBP voice rewrite error-path coverage verifies missing target embedded-LC state
 is logged with the handling router name and does not crash packet processing.
 HBP voice lifecycle coverage verifies a voice terminator marks the slot stream

docs/v1x-codex-changelog.md

@@ -81,6 +81,9 @@
 - Fixed outbound FBP `DMRE` version bytes to use the negotiated OpenBridge
   session version rather than the module default version constant. BCVE remains
   an advertisement of the local maximum supported version.
+- Made FBP/OpenBridge per-stream rate-drop protection more tolerant of short
+  trunk bursts, using a sustained over-rate threshold and logging peer/source
+  metadata. The OBP path no longer calls HBP proxy blacklist handling.
 - Removed obsolete top-level sample configs with invalid `PROTO_VER: 2`
   examples, and removed the old non-compose Docker installer. Docker Compose is
   the recommended Docker install path.

tests/test_deterministic_harness.py

@@ -2794,16 +2794,52 @@ DEFAULT_DIAL_TS2: 92
             self.assertIn("PACKET RATE 0.50/s", "\n".join(logs.output))
             self.assertNotIn("PACKET RATE 2.00/s", "\n".join(logs.output))
 
-    def test_obp_group_rate_drop_uses_elapsed_duration(self):
+    def test_obp_group_rate_drop_ignores_short_burst(self):
         config = minimal_config(("MASTER-A",))
         add_openbridge_system(config, "OBP-1", network_id=3001)
 
         with DeterministicScenario(config=config) as scenario:
-            rate_drops = []
-            scenario.systems["OBP-1"].proxy_BadPeer = lambda: rate_drops.append(True)
+            stream_id = 0x01020304
+            first_packet = PacketSpec(
+                dst_id=123,
+                slot=1,
+                stream_id=stream_id,
+                seq=0,
+                call_type="group",
+                frame_type=HBPF_VOICE,
+                dtype_vseq=0,
+                payload=bytes([0]) * 33,
+            )
+            scenario.inject_obp("OBP-1", first_packet)
+            scenario.clock.advance(0.1)
+
+            with self.assertNoLogs(scenario.bm.logger, level="WARNING"):
+                for seq in range(1, 61):
+                    packet = PacketSpec(
+                        dst_id=123,
+                        slot=1,
+                        stream_id=stream_id,
+                        seq=seq,
+                        call_type="group",
+                        frame_type=HBPF_VOICE,
+                        dtype_vseq=0,
+                        payload=bytes([seq]) * 33,
+                    )
+                    scenario.inject_obp("OBP-1", packet)
 
+            self.assertEqual(
+                scenario.systems["OBP-1"].STATUS[bytes_4(stream_id)]["packets"],
+                60,
+            )
+
+    def test_obp_group_rate_drop_uses_lenient_sustained_fbp_threshold(self):
+        config = minimal_config(("MASTER-A",))
+        add_openbridge_system(config, "OBP-1", network_id=3001)
+
+        with DeterministicScenario(config=config) as scenario:
             stream_id = 0x01020304
             first_packet = PacketSpec(
+                peer_id=3001,
                 dst_id=123,
                 slot=1,
                 stream_id=stream_id,
@@ -2814,11 +2850,12 @@ DEFAULT_DIAL_TS2: 92
                 payload=bytes([0]) * 33,
             )
             scenario.inject_obp("OBP-1", first_packet)
-            scenario.clock.advance(0.5)
+            scenario.clock.advance(2.0)
 
             with self.assertLogs(scenario.bm.logger, level="WARNING") as logs:
-                for seq in range(1, 20):
+                for seq in range(1, 102):
                     packet = PacketSpec(
+                        peer_id=3001,
                         dst_id=123,
                         slot=1,
                         stream_id=stream_id,
@@ -2830,8 +2867,11 @@ DEFAULT_DIAL_TS2: 92
                     )
                     scenario.inject_obp("OBP-1", packet)
 
-            self.assertEqual(rate_drops, [True])
-            self.assertIn("*PacketControl* RATE DROP!", "\n".join(logs.output))
+            log_output = "\n".join(logs.output)
+            self.assertIn("*PacketControl* RATE DROP!", log_output)
+            self.assertIn("PEER: 3001", log_output)
+            self.assertIn("SRC: 9990", log_output)
+            self.assertIn("DURATION: 2.00", log_output)
 
     def test_obp_voice_terminator_marks_stream_finished_when_reports_disabled(self):
         config = minimal_config(("MASTER-B",))

tests/test_udp_blackbox_harness.py

@@ -1218,6 +1218,50 @@ class UdpBlackBoxHarnessTest(unittest.TestCase):
         self.assertEqual(captured.fields["slot"], 2)
         self.assertEqual(captured.fields["stream_id"], bytes_4(0x01020305))
 
+    def test_fbp_sustained_over_rate_stream_is_dropped_without_proxy_error(self):
+        require_udp_integration_enabled()
+
+        stream_id = 0x01020316
+        with UdpBlackBoxScenario(fbp_systems={"OBP-1": 3001}) as scenario:
+            fbp_peer = scenario.fbp_peer("OBP-1")
+
+            fbp_peer.send_fbp(
+                PacketSpec(
+                    peer_id=3001,
+                    rf_src=3120001,
+                    dst_id=91,
+                    slot=1,
+                    stream_id=stream_id,
+                    seq=0,
+                    payload=bytes([0]) * 33,
+                ),
+                source_server=9991,
+                source_rptr=1001,
+            )
+            time.sleep(2.0)
+
+            for seq in range(1, 102):
+                fbp_peer.send_fbp(
+                    PacketSpec(
+                        peer_id=3001,
+                        rf_src=3120001,
+                        dst_id=91,
+                        slot=1,
+                        stream_id=stream_id,
+                        seq=seq,
+                        payload=bytes([seq]) * 33,
+                    ),
+                    source_server=9991,
+                    source_rptr=1001,
+                )
+
+            output = scenario.process.wait_for_log("*PacketControl* RATE DROP!", timeout=3.0)
+
+        self.assertIn("PEER: 3001", output)
+        self.assertIn("SRC: 9991", output)
+        self.assertNotIn("proxy_BadPeer", output)
+        self.assertNotIn("AttributeError", output)
+
     def test_fbp_source_quench_suppresses_hbp_to_fbp_stream(self):
         require_udp_integration_enabled()