diff --git a/bbot/scanner/scanner.py b/bbot/scanner/scanner.py
index c7a7aa0a1e..8b9d018df5 100644
--- a/bbot/scanner/scanner.py
+++ b/bbot/scanner/scanner.py
@@ -756,6 +756,14 @@ def modules_status(self, _log=False, detailed=False):
         mem_percent = mem_status.percent
         prev_delay = self._ingress_delay
         new_delay = self._compute_ingress_delay(mem_percent)
+        # if no module has work in flight or queued, ingress is the only drain — don't throttle it
+        drain_mode = not any(
+            (m.running or m.outgoing_event_queue.qsize() > 0 or m.num_incoming_events > 0)
+            for m in self.modules.values()
+            if not m._intercept
+        )
+        if drain_mode:
+            new_delay = 0.0
         self._ingress_delay = new_delay
         if mem_percent > self.max_mem_percent:
             free_memory_human = self.helpers.bytes_to_human(mem_status.available)
diff --git a/bbot/test/test_step_1/test_scan.py b/bbot/test/test_step_1/test_scan.py
index 45f97bd97d..d094fd4093 100644
--- a/bbot/test/test_step_1/test_scan.py
+++ b/bbot/test/test_step_1/test_scan.py
@@ -460,18 +460,63 @@ def mock_memory_status():
     # status loop wires memory_status -> _ingress_delay
     assert scan._ingress_delay == 0.0
 
-    mem_percent[0] = 93.0
-    scan.modules_status(_log=False)
-    assert scan._ingress_delay == pytest.approx(3.0), "delay should engage above threshold"
-
-    mem_percent[0] = 97.0
-    scan.modules_status(_log=False)
-    assert scan._ingress_delay == pytest.approx(5.0), "delay should clamp at the cap"
-
-    mem_percent[0] = 80.0
-    scan.modules_status(_log=False)
-    assert scan._ingress_delay == 0.0, "delay should clear once memory drops back"
+    # the drain-mode bypass only zeros the delay when no module has work; pin a fake task so
+    # the engagement curve below exercises the memory-driven formula, not the bypass
+    target_module = next(m for m in scan.modules.values() if not m._intercept)
+    target_module._task_counter.tasks["fake-task"] = SimpleNamespace(n=1)
+    try:
+        mem_percent[0] = 93.0
+        scan.modules_status(_log=False)
+        assert scan._ingress_delay == pytest.approx(3.0), "delay should engage above threshold"
+
+        mem_percent[0] = 97.0
+        scan.modules_status(_log=False)
+        assert scan._ingress_delay == pytest.approx(5.0), "delay should clamp at the cap"
+
+        mem_percent[0] = 80.0
+        scan.modules_status(_log=False)
+        assert scan._ingress_delay == 0.0, "delay should clear once memory drops back"
+    finally:
+        target_module._task_counter.tasks.pop("fake-task", None)
 
     # scan still produces events with the throttle disengaged
     events = [e async for e in scan.async_start()]
     assert any(e.type == "IP_ADDRESS" for e in events), "scan should still produce events"
+
+
+@pytest.mark.asyncio
+async def test_memory_backpressure_drain_mode_bypass(bbot_scanner, monkeypatch):
+    """Throttle must clear in drain mode: with memory pinned high but no module producing,
+    ingress IS the drain and throttling it traps the scan in a feedback loop."""
+    from types import SimpleNamespace
+
+    mem_percent = [93.0]
+
+    def mock_memory_status():
+        return SimpleNamespace(percent=mem_percent[0], available=1_000_000_000)
+
+    scan = bbot_scanner("127.0.0.1", config={"max_mem_percent": 90})
+    await scan._prep()
+    monkeypatch.setattr("bbot.core.helpers.misc.memory_status", mock_memory_status)
+
+    non_intercept = [m for m in scan.modules.values() if not m._intercept]
+    assert non_intercept, "test requires at least one non-intercept module"
+
+    # simulate a healthy scan: at least one module is mid-handle
+    target_module = non_intercept[0]
+    target_module._task_counter.tasks["fake-task"] = SimpleNamespace(n=1)
+    try:
+        scan.modules_status(_log=False)
+        assert scan._ingress_delay == pytest.approx(3.0), (
+            "throttle should engage when memory is high and a module is running"
+        )
+    finally:
+        target_module._task_counter.tasks.pop("fake-task", None)
+
+    # now drain mode: no module running, no module has queued work
+    for m in non_intercept:
+        assert not m.running
+        assert m.outgoing_event_queue.qsize() == 0
+        assert m.num_incoming_events == 0
+    scan.modules_status(_log=False)
+    assert scan._ingress_delay == 0.0, "throttle must clear in drain mode (no producers, ingress is the drain)"