diff --git a/src/core/automount.c b/src/core/automount.c
index 57d12657234..edc95881653 100644
--- a/src/core/automount.c
+++ b/src/core/automount.c
@@ -84,7 +84,7 @@ static void unmount_autofs(Automount *a) {
         if (a->pipe_fd < 0)
                 return;
 
-        a->pipe_event_source = sd_event_source_unref(a->pipe_event_source);
+        a->pipe_event_source = sd_event_source_disable_unref(a->pipe_event_source);
         a->pipe_fd = safe_close(a->pipe_fd);
 
         /* If we reload/reexecute things we keep the mount point around */
@@ -113,7 +113,7 @@ static void automount_done(Unit *u) {
         a->tokens = set_free(a->tokens);
         a->expire_tokens = set_free(a->expire_tokens);
 
-        a->expire_event_source = sd_event_source_unref(a->expire_event_source);
+        a->expire_event_source = sd_event_source_disable_unref(a->expire_event_source);
 }
 
 static int automount_add_trigger_dependencies(Automount *a) {
diff --git a/src/core/cgroup.c b/src/core/cgroup.c
index 3f8e06240fa..b44391ac22f 100644
--- a/src/core/cgroup.c
+++ b/src/core/cgroup.c
@@ -3200,7 +3200,7 @@ int manager_setup_cgroup(Manager *m) {
         }
 
         /* 3. Allocate cgroup empty defer event source */
-        m->cgroup_empty_event_source = sd_event_source_unref(m->cgroup_empty_event_source);
+        m->cgroup_empty_event_source = sd_event_source_disable_unref(m->cgroup_empty_event_source);
         r = sd_event_add_defer(m->event, &m->cgroup_empty_event_source, on_cgroup_empty_event, m);
         if (r < 0)
                 return log_error_errno(r, "Failed to create cgroup empty event source: %m");
@@ -3223,7 +3223,7 @@ int manager_setup_cgroup(Manager *m) {
 
                 /* In the unified hierarchy we can get cgroup empty notifications via inotify. */
 
-                m->cgroup_inotify_event_source = sd_event_source_unref(m->cgroup_inotify_event_source);
+                m->cgroup_inotify_event_source = sd_event_source_disable_unref(m->cgroup_inotify_event_source);
                 safe_close(m->cgroup_inotify_fd);
 
                 m->cgroup_inotify_fd = inotify_init1(IN_NONBLOCK|IN_CLOEXEC);
@@ -3306,12 +3306,12 @@ void manager_shutdown_cgroup(Manager *m, bool delete) {
         if (delete && m->cgroup_root && m->test_run_flags != MANAGER_TEST_RUN_MINIMAL)
                 (void) cg_trim(SYSTEMD_CGROUP_CONTROLLER, m->cgroup_root, false);
 
-        m->cgroup_empty_event_source = sd_event_source_unref(m->cgroup_empty_event_source);
+        m->cgroup_empty_event_source = sd_event_source_disable_unref(m->cgroup_empty_event_source);
 
         m->cgroup_control_inotify_wd_unit = hashmap_free(m->cgroup_control_inotify_wd_unit);
         m->cgroup_memory_inotify_wd_unit = hashmap_free(m->cgroup_memory_inotify_wd_unit);
 
-        m->cgroup_inotify_event_source = sd_event_source_unref(m->cgroup_inotify_event_source);
+        m->cgroup_inotify_event_source = sd_event_source_disable_unref(m->cgroup_inotify_event_source);
         m->cgroup_inotify_fd = safe_close(m->cgroup_inotify_fd);
 
         m->pin_cgroupfs_fd = safe_close(m->pin_cgroupfs_fd);
diff --git a/src/core/dbus.c b/src/core/dbus.c
index 26e34ac4d0f..f876433c00e 100644
--- a/src/core/dbus.c
+++ b/src/core/dbus.c
@@ -1037,7 +1037,7 @@ void bus_done_private(Manager *m) {
 
         m->private_buses = set_free(m->private_buses);
 
-        m->private_listen_event_source = sd_event_source_unref(m->private_listen_event_source);
+        m->private_listen_event_source = sd_event_source_disable_unref(m->private_listen_event_source);
         m->private_listen_fd = safe_close(m->private_listen_fd);
 }
 
diff --git a/src/core/job.c b/src/core/job.c
index a273e4220e2..afe0985ce18 100644
--- a/src/core/job.c
+++ b/src/core/job.c
@@ -88,7 +88,7 @@ void job_unlink(Job *j) {
                 j->in_gc_queue = false;
         }
 
-        j->timer_event_source = sd_event_source_unref(j->timer_event_source);
+        j->timer_event_source = sd_event_source_disable_unref(j->timer_event_source);
 }
 
 Job* job_free(Job *j) {
@@ -129,7 +129,7 @@ static void job_set_state(Job *j, JobState state) {
                 j->unit->manager->n_running_jobs--;
 
                 if (j->unit->manager->n_running_jobs <= 0)
-                        j->unit->manager->jobs_in_progress_event_source = sd_event_source_unref(j->unit->manager->jobs_in_progress_event_source);
+                        j->unit->manager->jobs_in_progress_event_source = sd_event_source_disable_unref(j->unit->manager->jobs_in_progress_event_source);
         }
 }
 
@@ -1336,7 +1336,7 @@ int job_coldplug(Job *j) {
         if (timeout_time == USEC_INFINITY)
                 return 0;
 
-        j->timer_event_source = sd_event_source_unref(j->timer_event_source);
+        j->timer_event_source = sd_event_source_disable_unref(j->timer_event_source);
 
         r = sd_event_add_time(
                         j->manager->event,
diff --git a/src/core/manager.c b/src/core/manager.c
index f3275a4070e..f8676be422f 100644
--- a/src/core/manager.c
+++ b/src/core/manager.c
@@ -287,7 +287,7 @@ static int manager_dispatch_ask_password_fd(sd_event_source *source,
 static void manager_close_ask_password(Manager *m) {
         assert(m);
 
-        m->ask_password_event_source = sd_event_source_unref(m->ask_password_event_source);
+        m->ask_password_event_source = sd_event_source_disable_unref(m->ask_password_event_source);
         m->ask_password_inotify_fd = safe_close(m->ask_password_inotify_fd);
         m->have_ask_password = -EINVAL;
 }
@@ -356,7 +356,7 @@ static int manager_watch_idle_pipe(Manager *m) {
 static void manager_close_idle_pipe(Manager *m) {
         assert(m);
 
-        m->idle_pipe_event_source = sd_event_source_unref(m->idle_pipe_event_source);
+        m->idle_pipe_event_source = sd_event_source_disable_unref(m->idle_pipe_event_source);
 
         safe_close_pair(m->idle_pipe);
         safe_close_pair(m->idle_pipe + 2);
@@ -370,7 +370,7 @@ static int manager_setup_time_change(Manager *m) {
         if (MANAGER_IS_TEST_RUN(m))
                 return 0;
 
-        m->time_change_event_source = sd_event_source_unref(m->time_change_event_source);
+        m->time_change_event_source = sd_event_source_disable_unref(m->time_change_event_source);
         m->time_change_fd = safe_close(m->time_change_fd);
 
         m->time_change_fd = time_change_fd();
@@ -938,7 +938,7 @@ static int manager_setup_notify(Manager *m) {
 
                 /* First free all secondary fields */
                 m->notify_socket = mfree(m->notify_socket);
-                m->notify_event_source = sd_event_source_unref(m->notify_event_source);
+                m->notify_event_source = sd_event_source_disable_unref(m->notify_event_source);
 
                 fd = socket(AF_UNIX, SOCK_DGRAM|SOCK_CLOEXEC|SOCK_NONBLOCK, 0);
                 if (fd < 0)
@@ -1026,7 +1026,7 @@ static int manager_setup_cgroups_agent(Manager *m) {
                 _cleanup_close_ int fd = -1;
 
                 /* First free all secondary fields */
-                m->cgroups_agent_event_source = sd_event_source_unref(m->cgroups_agent_event_source);
+                m->cgroups_agent_event_source = sd_event_source_disable_unref(m->cgroups_agent_event_source);
 
                 fd = socket(AF_UNIX, SOCK_DGRAM|SOCK_CLOEXEC|SOCK_NONBLOCK, 0);
                 if (fd < 0)
@@ -1091,7 +1091,7 @@ static int manager_setup_user_lookup_fd(Manager *m) {
 
                 /* Free all secondary fields */
                 safe_close_pair(m->user_lookup_fds);
-                m->user_lookup_event_source = sd_event_source_unref(m->user_lookup_event_source);
+                m->user_lookup_event_source = sd_event_source_disable_unref(m->user_lookup_event_source);
 
                 if (socketpair(AF_UNIX, SOCK_DGRAM|SOCK_CLOEXEC, 0, m->user_lookup_fds) < 0)
                         return log_error_errno(errno, "Failed to allocate user lookup socket: %m");
@@ -3689,7 +3689,7 @@ int manager_deserialize(Manager *m, FILE *f, FDSet *fds) {
                         if (safe_atoi(val, &fd) < 0 || fd < 0 || !fdset_contains(fds, fd))
                                 log_notice("Failed to parse notify fd, ignoring: \"%s\"", val);
                         else {
-                                m->notify_event_source = sd_event_source_unref(m->notify_event_source);
+                                m->notify_event_source = sd_event_source_disable_unref(m->notify_event_source);
                                 safe_close(m->notify_fd);
                                 m->notify_fd = fdset_remove(fds, fd);
                         }
@@ -3705,7 +3705,7 @@ int manager_deserialize(Manager *m, FILE *f, FDSet *fds) {
                         if (safe_atoi(val, &fd) < 0 || fd < 0 || !fdset_contains(fds, fd))
                                 log_notice("Failed to parse cgroups agent fd, ignoring.: %s", val);
                         else {
-                                m->cgroups_agent_event_source = sd_event_source_unref(m->cgroups_agent_event_source);
+                                m->cgroups_agent_event_source = sd_event_source_disable_unref(m->cgroups_agent_event_source);
                                 safe_close(m->cgroups_agent_fd);
                                 m->cgroups_agent_fd = fdset_remove(fds, fd);
                         }
@@ -3716,7 +3716,7 @@ int manager_deserialize(Manager *m, FILE *f, FDSet *fds) {
                         if (sscanf(val, "%i %i", &fd0, &fd1) != 2 || fd0 < 0 || fd1 < 0 || fd0 == fd1 || !fdset_contains(fds, fd0) || !fdset_contains(fds, fd1))
                                 log_notice("Failed to parse user lookup fd, ignoring: %s", val);
                         else {
-                                m->user_lookup_event_source = sd_event_source_unref(m->user_lookup_event_source);
+                                m->user_lookup_event_source = sd_event_source_disable_unref(m->user_lookup_event_source);
                                 safe_close_pair(m->user_lookup_fds);
                                 m->user_lookup_fds[0] = fdset_remove(fds, fd0);
                                 m->user_lookup_fds[1] = fdset_remove(fds, fd1);
diff --git a/src/core/mount.c b/src/core/mount.c
index ee4dd0eea19..053deac14db 100644
--- a/src/core/mount.c
+++ b/src/core/mount.c
@@ -266,7 +266,7 @@ static void mount_done(Unit *u) {
 
         mount_unwatch_control_pid(m);
 
-        m->timer_event_source = sd_event_source_unref(m->timer_event_source);
+        m->timer_event_source = sd_event_source_disable_unref(m->timer_event_source);
 }
 
 static int update_parameters_proc_self_mountinfo(
@@ -718,7 +718,7 @@ static void mount_set_state(Mount *m, MountState state) {
         m->state = state;
 
         if (!MOUNT_STATE_WITH_PROCESS(state)) {
-                m->timer_event_source = sd_event_source_unref(m->timer_event_source);
+                m->timer_event_source = sd_event_source_disable_unref(m->timer_event_source);
                 mount_unwatch_control_pid(m);
                 m->control_command = NULL;
                 m->control_command_id = _MOUNT_EXEC_COMMAND_INVALID;
@@ -1787,7 +1787,7 @@ static int mount_load_proc_self_mountinfo(Manager *m, bool set_flags) {
 static void mount_shutdown(Manager *m) {
         assert(m);
 
-        m->mount_event_source = sd_event_source_unref(m->mount_event_source);
+        m->mount_event_source = sd_event_source_disable_unref(m->mount_event_source);
 
         mnt_unref_monitor(m->mount_monitor);
         m->mount_monitor = NULL;
@@ -2127,7 +2127,7 @@ static int mount_clean(Unit *u, ExecCleanMask mask) {
 fail:
         log_unit_warning_errno(u, r, "Failed to initiate cleaning: %m");
         m->clean_result = MOUNT_FAILURE_RESOURCES;
-        m->timer_event_source = sd_event_source_unref(m->timer_event_source);
+        m->timer_event_source = sd_event_source_disable_unref(m->timer_event_source);
         return r;
 }
 
diff --git a/src/core/path.c b/src/core/path.c
index d27e923ace9..e098e83a311 100644
--- a/src/core/path.c
+++ b/src/core/path.c
@@ -167,7 +167,7 @@ fail:
 void path_spec_unwatch(PathSpec *s) {
         assert(s);
 
-        s->event_source = sd_event_source_unref(s->event_source);
+        s->event_source = sd_event_source_disable_unref(s->event_source);
         s->inotify_fd = safe_close(s->inotify_fd);
 }
 
diff --git a/src/core/scope.c b/src/core/scope.c
index a247da206ff..af6311bb5f6 100644
--- a/src/core/scope.c
+++ b/src/core/scope.c
@@ -48,7 +48,7 @@ static void scope_done(Unit *u) {
         s->controller = mfree(s->controller);
         s->controller_track = sd_bus_track_unref(s->controller_track);
 
-        s->timer_event_source = sd_event_source_unref(s->timer_event_source);
+        s->timer_event_source = sd_event_source_disable_unref(s->timer_event_source);
 }
 
 static int scope_arm_timer(Scope *s, usec_t usec) {
@@ -92,7 +92,7 @@ static void scope_set_state(Scope *s, ScopeState state) {
         s->state = state;
 
         if (!IN_SET(state, SCOPE_STOP_SIGTERM, SCOPE_STOP_SIGKILL))
-                s->timer_event_source = sd_event_source_unref(s->timer_event_source);
+                s->timer_event_source = sd_event_source_disable_unref(s->timer_event_source);
 
         if (IN_SET(state, SCOPE_DEAD, SCOPE_FAILED)) {
                 unit_unwatch_all_pids(UNIT(s));
diff --git a/src/core/service.c b/src/core/service.c
index f13747c2936..0b7555c856e 100644
--- a/src/core/service.c
+++ b/src/core/service.c
@@ -196,7 +196,7 @@ void service_close_socket_fd(Service *s) {
 static void service_stop_watchdog(Service *s) {
         assert(s);
 
-        s->watchdog_event_source = sd_event_source_unref(s->watchdog_event_source);
+        s->watchdog_event_source = sd_event_source_disable_unref(s->watchdog_event_source);
         s->watchdog_timestamp = DUAL_TIMESTAMP_NULL;
 }
 
@@ -395,8 +395,8 @@ static void service_done(Unit *u) {
 
         service_stop_watchdog(s);
 
-        s->timer_event_source = sd_event_source_unref(s->timer_event_source);
-        s->exec_fd_event_source = sd_event_source_unref(s->exec_fd_event_source);
+        s->timer_event_source = sd_event_source_disable_unref(s->timer_event_source);
+        s->exec_fd_event_source = sd_event_source_disable_unref(s->exec_fd_event_source);
 
         service_release_resources(u);
 }
@@ -1055,7 +1055,7 @@ static void service_set_state(Service *s, ServiceState state) {
                     SERVICE_FINAL_WATCHDOG, SERVICE_FINAL_SIGTERM, SERVICE_FINAL_SIGKILL,
                     SERVICE_AUTO_RESTART,
                     SERVICE_CLEANING))
-                s->timer_event_source = sd_event_source_unref(s->timer_event_source);
+                s->timer_event_source = sd_event_source_disable_unref(s->timer_event_source);
 
         if (!IN_SET(state,
                     SERVICE_START, SERVICE_START_POST,
@@ -1091,7 +1091,7 @@ static void service_set_state(Service *s, ServiceState state) {
                 service_close_socket_fd(s);
 
         if (state != SERVICE_START)
-                s->exec_fd_event_source = sd_event_source_unref(s->exec_fd_event_source);
+                s->exec_fd_event_source = sd_event_source_disable_unref(s->exec_fd_event_source);
 
         if (!IN_SET(state, SERVICE_START_POST, SERVICE_RUNNING, SERVICE_RELOAD))
                 service_stop_watchdog(s);
@@ -1351,7 +1351,7 @@ static int service_allocate_exec_fd_event_source(
         if (r < 0)
                 return log_unit_error_errno(UNIT(s), r, "Failed to adjust priority of exec_fd event source: %m");
 
-        (void) sd_event_source_set_description(source, "service event_fd");
+        (void) sd_event_source_set_description(source, "service exec_fd");
 
         r = sd_event_source_set_io_fd_own(source, true);
         if (r < 0)
@@ -1433,6 +1433,8 @@ static int service_spawn(
         if (r < 0)
                 return r;
 
+        assert(!s->exec_fd_event_source);
+
         if (flags & EXEC_IS_CONTROL) {
                 /* If this is a control process, mask the permissions/chroot application if this is requested. */
                 if (s->permissions_start_only)
@@ -1458,8 +1460,6 @@ static int service_spawn(
         }
 
         if (!FLAGS_SET(flags, EXEC_IS_CONTROL) && s->type == SERVICE_EXEC) {
-                assert(!s->exec_fd_event_source);
-
                 r = service_allocate_exec_fd(s, &exec_fd_source, &exec_params.exec_fd);
                 if (r < 0)
                         return r;
@@ -3019,7 +3019,7 @@ static int service_deserialize_item(Unit *u, const char *key, const char *value,
                 if (safe_atoi(value, &fd) < 0 || fd < 0 || !fdset_contains(fds, fd))
                         log_unit_debug(u, "Failed to parse exec-fd value: %s", value);
                 else {
-                        s->exec_fd_event_source = sd_event_source_unref(s->exec_fd_event_source);
+                        s->exec_fd_event_source = sd_event_source_disable_unref(s->exec_fd_event_source);
 
                         fd = fdset_remove(fds, fd);
                         if (service_allocate_exec_fd_event_source(s, fd, &s->exec_fd_event_source) < 0)
@@ -3216,7 +3216,7 @@ static int service_dispatch_exec_io(sd_event_source *source, int fd, uint32_t ev
                 }
                 if (n == 0) { /* EOF → the event we are waiting for */
 
-                        s->exec_fd_event_source = sd_event_source_unref(s->exec_fd_event_source);
+                        s->exec_fd_event_source = sd_event_source_disable_unref(s->exec_fd_event_source);
 
                         if (s->exec_fd_hot) { /* Did the child tell us to expect EOF now? */
                                 log_unit_debug(UNIT(s), "Got EOF on exec-fd");
@@ -3384,6 +3384,11 @@ static void service_sigchld_event(Unit *u, pid_t pid, int code, int status) {
         else
                 clean_mode = EXIT_CLEAN_DAEMON;
 
+        if (s->main_pid == pid)
+                /* Clean up the exec_fd event source. The source owns its end of the pipe, so this will close
+                 * that too. */
+                s->exec_fd_event_source = sd_event_source_disable_unref(s->exec_fd_event_source);
+
         if (is_clean_exit(code, status, clean_mode, &s->success_status))
                 f = SERVICE_SUCCESS;
         else if (code == CLD_EXITED)
@@ -4415,7 +4420,7 @@ static int service_clean(Unit *u, ExecCleanMask mask) {
 fail:
         log_unit_warning_errno(u, r, "Failed to initiate cleaning: %m");
         s->clean_result = SERVICE_FAILURE_RESOURCES;
-        s->timer_event_source = sd_event_source_unref(s->timer_event_source);
+        s->timer_event_source = sd_event_source_disable_unref(s->timer_event_source);
         return r;
 }
 
diff --git a/src/core/socket.c b/src/core/socket.c
index a64bf7360cd..d93186ddbd1 100644
--- a/src/core/socket.c
+++ b/src/core/socket.c
@@ -174,7 +174,7 @@ static void socket_done(Unit *u) {
 
         s->fdname = mfree(s->fdname);
 
-        s->timer_event_source = sd_event_source_unref(s->timer_event_source);
+        s->timer_event_source = sd_event_source_disable_unref(s->timer_event_source);
 }
 
 static int socket_arm_timer(Socket *s, usec_t usec) {
@@ -934,7 +934,7 @@ static void socket_close_fds(Socket *s) {
 
                 was_open = p->fd >= 0;
 
-                p->event_source = sd_event_source_unref(p->event_source);
+                p->event_source = sd_event_source_disable_unref(p->event_source);
                 p->fd = safe_close(p->fd);
                 socket_cleanup_fd_list(p);
 
@@ -1834,7 +1834,7 @@ static void socket_set_state(Socket *s, SocketState state) {
                     SOCKET_FINAL_SIGKILL,
                     SOCKET_CLEANING)) {
 
-                s->timer_event_source = sd_event_source_unref(s->timer_event_source);
+                s->timer_event_source = sd_event_source_disable_unref(s->timer_event_source);
                 socket_unwatch_control_pid(s);
                 s->control_command = NULL;
                 s->control_command_id = _SOCKET_EXEC_COMMAND_INVALID;
@@ -2048,7 +2048,7 @@ static int socket_chown(Socket *s, pid_t *_pid) {
         return 0;
 
 fail:
-        s->timer_event_source = sd_event_source_unref(s->timer_event_source);
+        s->timer_event_source = sd_event_source_disable_unref(s->timer_event_source);
         return r;
 }
 
@@ -3416,7 +3416,7 @@ static int socket_clean(Unit *u, ExecCleanMask mask) {
 fail:
         log_unit_warning_errno(u, r, "Failed to initiate cleaning: %m");
         s->clean_result = SOCKET_FAILURE_RESOURCES;
-        s->timer_event_source = sd_event_source_unref(s->timer_event_source);
+        s->timer_event_source = sd_event_source_disable_unref(s->timer_event_source);
         return r;
 }
 
diff --git a/src/core/swap.c b/src/core/swap.c
index 8aecc391e7b..3843b195000 100644
--- a/src/core/swap.c
+++ b/src/core/swap.c
@@ -179,7 +179,7 @@ static void swap_done(Unit *u) {
 
         swap_unwatch_control_pid(s);
 
-        s->timer_event_source = sd_event_source_unref(s->timer_event_source);
+        s->timer_event_source = sd_event_source_disable_unref(s->timer_event_source);
 }
 
 static int swap_arm_timer(Swap *s, usec_t usec) {
@@ -549,7 +549,7 @@ static void swap_set_state(Swap *s, SwapState state) {
         s->state = state;
 
         if (!SWAP_STATE_WITH_PROCESS(state)) {
-                s->timer_event_source = sd_event_source_unref(s->timer_event_source);
+                s->timer_event_source = sd_event_source_disable_unref(s->timer_event_source);
                 swap_unwatch_control_pid(s);
                 s->control_command = NULL;
                 s->control_command_id = _SWAP_EXEC_COMMAND_INVALID;
@@ -710,7 +710,7 @@ static int swap_spawn(Swap *s, ExecCommand *c, pid_t *_pid) {
         return 0;
 
 fail:
-        s->timer_event_source = sd_event_source_unref(s->timer_event_source);
+        s->timer_event_source = sd_event_source_disable_unref(s->timer_event_source);
 
         return r;
 }
@@ -1368,7 +1368,7 @@ static int swap_following_set(Unit *u, Set **_set) {
 static void swap_shutdown(Manager *m) {
         assert(m);
 
-        m->swap_event_source = sd_event_source_unref(m->swap_event_source);
+        m->swap_event_source = sd_event_source_disable_unref(m->swap_event_source);
         m->proc_swaps = safe_fclose(m->proc_swaps);
         m->swaps_by_devnode = hashmap_free(m->swaps_by_devnode);
 }
@@ -1576,7 +1576,7 @@ static int swap_clean(Unit *u, ExecCleanMask mask) {
 fail:
         log_unit_warning_errno(u, r, "Failed to initiate cleaning: %m");
         s->clean_result = SWAP_FAILURE_RESOURCES;
-        s->timer_event_source = sd_event_source_unref(s->timer_event_source);
+        s->timer_event_source = sd_event_source_disable_unref(s->timer_event_source);
         return r;
 }
 
diff --git a/src/core/timer.c b/src/core/timer.c
index b0caaf38504..e064ad9a2dc 100644
--- a/src/core/timer.c
+++ b/src/core/timer.c
@@ -65,8 +65,8 @@ static void timer_done(Unit *u) {
 
         timer_free_values(t);
 
-        t->monotonic_event_source = sd_event_source_unref(t->monotonic_event_source);
-        t->realtime_event_source = sd_event_source_unref(t->realtime_event_source);
+        t->monotonic_event_source = sd_event_source_disable_unref(t->monotonic_event_source);
+        t->realtime_event_source = sd_event_source_disable_unref(t->realtime_event_source);
 
         free(t->stamp_path);
 }
@@ -296,8 +296,8 @@ static void timer_set_state(Timer *t, TimerState state) {
         t->state = state;
 
         if (state != TIMER_WAITING) {
-                t->monotonic_event_source = sd_event_source_unref(t->monotonic_event_source);
-                t->realtime_event_source = sd_event_source_unref(t->realtime_event_source);
+                t->monotonic_event_source = sd_event_source_disable_unref(t->monotonic_event_source);
+                t->realtime_event_source = sd_event_source_disable_unref(t->realtime_event_source);
                 t->next_elapse_monotonic_or_boottime = USEC_INFINITY;
                 t->next_elapse_realtime = USEC_INFINITY;
         }
diff --git a/src/core/unit.c b/src/core/unit.c
index be8e0ddef20..b203f16e9b2 100644
--- a/src/core/unit.c
+++ b/src/core/unit.c
@@ -2928,7 +2928,7 @@ void unit_dequeue_rewatch_pids(Unit *u) {
         if (r < 0)
                 log_warning_errno(r, "Failed to disable event source for tidying watched PIDs, ignoring: %m");
 
-        u->rewatch_pids_event_source = sd_event_source_unref(u->rewatch_pids_event_source);
+        u->rewatch_pids_event_source = sd_event_source_disable_unref(u->rewatch_pids_event_source);
 }
 
 bool unit_job_is_applicable(Unit *u, JobType j) {
diff --git a/src/libsystemd/sd-event/sd-event.c b/src/libsystemd/sd-event/sd-event.c
index cc9d9f5d9bc..e9199deb418 100644
--- a/src/libsystemd/sd-event/sd-event.c
+++ b/src/libsystemd/sd-event/sd-event.c
@@ -85,6 +85,11 @@ DEFINE_PRIVATE_STRING_TABLE_LOOKUP_TO_STRING(event_source_type, int);
                SOURCE_DEFER,                    \
                SOURCE_INOTIFY)
 
+/* This is used to assert that we didn't pass an unexpected source type to event_source_time_prioq_put().
+ * Time sources and ratelimited sources can be passed, so effectively this is the same as the
+ * EVENT_SOURCE_CAN_RATE_LIMIT() macro. */
+#define EVENT_SOURCE_USES_TIME_PRIOQ(t) EVENT_SOURCE_CAN_RATE_LIMIT(t)
+
 struct sd_event {
         unsigned n_ref;
 
@@ -1198,6 +1203,7 @@ static int event_source_time_prioq_put(
 
         assert(s);
         assert(d);
+        assert(EVENT_SOURCE_USES_TIME_PRIOQ(s->type));
 
         r = prioq_put(d->earliest, s, &s->earliest_index);
         if (r < 0)
@@ -2974,6 +2980,7 @@ static int event_arm_timer(
         d->needs_rearm = false;
 
         a = prioq_peek(d->earliest);
+        assert(!a || EVENT_SOURCE_USES_TIME_PRIOQ(a->type));
         if (!a || a->enabled == SD_EVENT_OFF || time_event_source_next(a) == USEC_INFINITY) {
 
                 if (d->fd < 0)
@@ -2991,7 +2998,8 @@ static int event_arm_timer(
         }
 
         b = prioq_peek(d->latest);
-        assert_se(b && b->enabled != SD_EVENT_OFF);
+        assert(!b || EVENT_SOURCE_USES_TIME_PRIOQ(b->type));
+        assert(b && b->enabled != SD_EVENT_OFF);
 
         t = sleep_between(e, time_event_source_next(a), time_event_source_latest(b));
         if (d->next == t)
@@ -3071,6 +3079,8 @@ static int process_timer(
 
         for (;;) {
                 s = prioq_peek(d->earliest);
+                assert(!s || EVENT_SOURCE_USES_TIME_PRIOQ(s->type));
+
                 if (!s || time_event_source_next(s) > n)
                         break;
 
@@ -3631,6 +3641,8 @@ static int dispatch_exit(sd_event *e) {
         assert(e);
 
         p = prioq_peek(e->exit);
+        assert(!p || p->type == SOURCE_EXIT);
+
         if (!p || event_source_is_offline(p)) {
                 e->state = SD_EVENT_FINISHED;
                 return 0;