diff --git a/src/libsystemd/sd-event/sd-event.c b/src/libsystemd/sd-event/sd-event.c
index 611af45293..489878a3e9 100644
--- a/src/libsystemd/sd-event/sd-event.c
+++ b/src/libsystemd/sd-event/sd-event.c
@@ -237,25 +237,6 @@ static usec_t time_event_source_next(const sd_event_source *s) {
         return USEC_INFINITY;
 }
 
-static int earliest_time_prioq_compare(const void *a, const void *b) {
-        const sd_event_source *x = a, *y = b;
-
-        /* Enabled ones first */
-        if (x->enabled != SD_EVENT_OFF && y->enabled == SD_EVENT_OFF)
-                return -1;
-        if (x->enabled == SD_EVENT_OFF && y->enabled != SD_EVENT_OFF)
-                return 1;
-
-        /* Move the pending ones to the end */
-        if (!x->pending && y->pending)
-                return -1;
-        if (x->pending && !y->pending)
-                return 1;
-
-        /* Order by time */
-        return CMP(time_event_source_next(x), time_event_source_next(y));
-}
-
 static usec_t time_event_source_latest(const sd_event_source *s) {
         assert(s);
 
@@ -274,7 +255,15 @@ static usec_t time_event_source_latest(const sd_event_source *s) {
         return USEC_INFINITY;
 }
 
-static int latest_time_prioq_compare(const void *a, const void *b) {
+static bool event_source_timer_candidate(const sd_event_source *s) {
+        assert(s);
+
+        /* Returns true for event sources that either are not pending yet (i.e. where it's worth to mark them pending)
+         * or which are currently ratelimited (i.e. where it's worth leaving the ratelimited state) */
+        return !s->pending || s->ratelimited;
+}
+
+static int time_prioq_compare(const void *a, const void *b, usec_t (*time_func)(const sd_event_source *s)) {
         const sd_event_source *x = a, *y = b;
 
         /* Enabled ones first */
@@ -283,14 +272,22 @@ static int latest_time_prioq_compare(const void *a, const void *b) {
         if (x->enabled == SD_EVENT_OFF && y->enabled != SD_EVENT_OFF)
                 return 1;
 
-        /* Move the pending ones to the end */
-        if (!x->pending && y->pending)
+        /* Order "non-pending OR ratelimited" before "pending AND not-ratelimited" */
+        if (event_source_timer_candidate(x) && !event_source_timer_candidate(y))
                 return -1;
-        if (x->pending && !y->pending)
+        if (!event_source_timer_candidate(x) && event_source_timer_candidate(y))
                 return 1;
 
         /* Order by time */
-        return CMP(time_event_source_latest(x), time_event_source_latest(y));
+        return CMP(time_func(x), time_func(y));
+}
+
+static int earliest_time_prioq_compare(const void *a, const void *b) {
+        return time_prioq_compare(a, b, time_event_source_next);
+}
+
+static int latest_time_prioq_compare(const void *a, const void *b) {
+        return time_prioq_compare(a, b, time_event_source_latest);
 }
 
 static int exit_prioq_compare(const void *a, const void *b) {
diff --git a/test/TEST-60-MOUNT-RATELIMIT/Makefile b/test/TEST-60-MOUNT-RATELIMIT/Makefile
new file mode 120000
index 0000000000..e9f93b1104
--- /dev/null
+++ b/test/TEST-60-MOUNT-RATELIMIT/Makefile
@@ -0,0 +1 @@
+../TEST-01-BASIC/Makefile
\ No newline at end of file
diff --git a/test/TEST-60-MOUNT-RATELIMIT/test.sh b/test/TEST-60-MOUNT-RATELIMIT/test.sh
new file mode 100755
index 0000000000..f9eb11ccb4
--- /dev/null
+++ b/test/TEST-60-MOUNT-RATELIMIT/test.sh
@@ -0,0 +1,7 @@
+#!/usr/bin/env bash
+set -e
+TEST_DESCRIPTION="Test that mount/unmount storms can enter/exit rate limit state and will not leak units"
+
+. $TEST_BASE_DIR/test-functions
+
+do_test "$@"
diff --git a/test/units/testsuite-60.service b/test/units/testsuite-60.service
new file mode 100644
index 0000000000..3715c4f341
--- /dev/null
+++ b/test/units/testsuite-60.service
@@ -0,0 +1,6 @@
+[Unit]
+Description=TEST-60-MOUNT-RATELIMIT
+
+[Service]
+Type=oneshot
+ExecStart=/usr/lib/systemd/tests/testdata/units/%N.sh
diff --git a/test/units/testsuite-60.sh b/test/units/testsuite-60.sh
new file mode 100755
index 0000000000..8158754667
--- /dev/null
+++ b/test/units/testsuite-60.sh
@@ -0,0 +1,73 @@
+#!/usr/bin/env bash
+set -eux
+set -o pipefail
+
+systemd-analyze log-level debug
+systemd-analyze log-target journal
+
+NUM_DIRS=20
+
+# mount/unmount enough times to trigger the /proc/self/mountinfo parsing rate limiting
+
+for ((i = 0; i < NUM_DIRS; i++)); do
+    mkdir "/tmp/meow${i}"
+done
+
+for ((i = 0; i < NUM_DIRS; i++)); do
+    mount -t tmpfs tmpfs "/tmp/meow${i}"
+done
+
+systemctl daemon-reload
+systemctl list-units -t mount tmp-meow* | grep -q tmp-meow
+
+for ((i = 0; i < NUM_DIRS; i++)); do
+    umount "/tmp/meow${i}"
+done
+
+# figure out if we have entered the rate limit state
+
+exited_rl=0
+timeout="$(date -ud "2 minutes" +%s)"
+while [[ $(date -u +%s) -le ${timeout} ]]; do
+    if journalctl -u init.scope | grep -q "(mount-monitor-dispatch) entered rate limit"; then
+        entered_rl=1
+        break
+    fi
+    sleep 5
+done
+
+# if the infra is slow we might not enter the rate limit state; in that case skip the exit check
+
+if [ "${entered_rl}" = "1" ]; then
+    exited_rl=0
+    timeout="$(date -ud "2 minutes" +%s)"
+    while [[ $(date -u +%s) -le ${timeout} ]]; do
+        if journalctl -u init.scope | grep -q "(mount-monitor-dispatch) left rate limit"; then
+            exited_rl=1
+            break
+        fi
+        sleep 5
+    done
+
+    if [ "${exited_rl}" = "0" ]; then
+        exit 24
+    fi
+fi
+
+# give some time for units to settle so we don't race between exiting the rate limit state and cleaning up the units
+
+sleep 60
+systemctl daemon-reload
+sleep 60
+
+# verify that the mount units are always cleaned up at the end
+
+if systemctl list-units -t mount tmp-meow* | grep -q tmp-meow; then
+    exit 42
+fi
+
+systemd-analyze log-level info
+
+echo OK >/testok
+
+exit 0