diff --git a/can/bus.py b/can/bus.py
index 555389b0f..af517e9d3 100644
--- a/can/bus.py
+++ b/can/bus.py
@@ -120,24 +120,18 @@ def recv(self, timeout: Optional[float] = None) -> Optional[Message]:
             # try to get a message
             msg, already_filtered = self._recv_internal(timeout=time_left)
 
+            # propagate timeouts from _recv_internal()
+            if not msg:
+                return None
+
             # return it, if it matches
-            if msg and (already_filtered or self._matches_filters(msg)):
+            if already_filtered or self._matches_filters(msg):
                 LOG.log(self.RECV_LOGGING_LEVEL, "Received: %s", msg)
                 return msg
 
-            # if not, and timeout is None, try indefinitely
-            elif timeout is None:
-                continue
-
-            # try next one only if there still is time, and with
-            # reduced timeout
-            else:
-                time_left = timeout - (time() - start)
-
-                if time_left > 0:
-                    continue
-
-                return None
+            # try again with reduced timeout
+            if timeout is not None:
+                time_left = max(0, timeout - (time() - start))
 
     def _recv_internal(
         self, timeout: Optional[float]
diff --git a/test/test_message_filtering.py b/test/test_message_filtering.py
index a73e07aa2..176257606 100644
--- a/test/test_message_filtering.py
+++ b/test/test_message_filtering.py
@@ -5,6 +5,7 @@
 """
 
 import unittest
+from unittest.mock import MagicMock, patch
 
 from can import Bus, Message
 
@@ -46,6 +47,57 @@ def test_match_example_message(self):
         self.assertFalse(self.bus._matches_filters(EXAMPLE_MSG))
         self.assertTrue(self.bus._matches_filters(HIGHEST_MSG))
 
+    def test_empty_queue_up_to_match(self):
+        bus2 = Bus(interface="virtual", channel="testy")
+        self.bus.set_filters(MATCH_EXAMPLE)
+        bus2.send(HIGHEST_MSG)
+        bus2.send(EXAMPLE_MSG)
+        actual = self.bus.recv(timeout=0)
+        bus2.shutdown()
+        self.assertTrue(
+            EXAMPLE_MSG.equals(
+                actual, timestamp_delta=None, check_direction=False, check_channel=False
+            )
+        )
+
+
+@patch("can.bus.time")
+class TestMessageFilterRetryTiming(unittest.TestCase):
+    def setUp(self):
+        self.bus = Bus(interface="virtual", channel="testy")
+        self.bus._recv_internal = MagicMock(name="_recv_internal")
+
+    def tearDown(self):
+        self.bus.shutdown()
+
+    def test_propagate_recv_internal_timeout(self, time: MagicMock) -> None:
+        self.bus._recv_internal.side_effect = [
+            (None, False),
+        ]
+        time.side_effect = [0]
+        self.bus.set_filters(MATCH_EXAMPLE)
+        self.assertIsNone(self.bus.recv(timeout=3))
+
+    def test_retry_with_adjusted_timeout(self, time: MagicMock) -> None:
+        self.bus._recv_internal.side_effect = [
+            (HIGHEST_MSG, False),
+            (EXAMPLE_MSG, False),
+        ]
+        time.side_effect = [0, 1]
+        self.bus.set_filters(MATCH_EXAMPLE)
+        self.bus.recv(timeout=3)
+        self.bus._recv_internal.assert_called_with(timeout=2)
+
+    def test_keep_timeout_non_negative(self, time: MagicMock) -> None:
+        self.bus._recv_internal.side_effect = [
+            (HIGHEST_MSG, False),
+            (EXAMPLE_MSG, False),
+        ]
+        time.side_effect = [0, 1]
+        self.bus.set_filters(MATCH_EXAMPLE)
+        self.bus.recv(timeout=0.5)
+        self.bus._recv_internal.assert_called_with(timeout=0)
+
 
 if __name__ == "__main__":
     unittest.main()