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#

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()