sd-future: rework cancellation, add slot callbacks and future groups

We replace the implicit single callback and wait-future indirection with
the usual slot design so independent fibers and future groups can track
the same future. We also opt to always have callbacks dispatch asynchronously,
including when they are registered after the future has already resolved.

We now require futures to be resolved before their final unref rather than
cancelling them implicitly. The idea is to catch bugs rather than leak futures.
A future should always be cancelled and awaited before it is cleaned up,
otherwise resources might leak if the stack is not unwound properly. We make
exceptions for futures that haven't started executing yet as there's no risk
of leaking resources there since there's no stack to unwind.

One issue with awaiting futures is that sd_fiber_await() returned both
errors and the future's result via the function's return value, meaning it
was impossible to distinguish future failures from interrupted waits (or
waits that failed for whatever other reason). This is generally not an issue
except in the cleanup functions, where we have to retry on interrupted waits
(the best example is a parent fiber waiting for a future being cancelled
itself, which should cancel the future it's waiting for and retry the wait).
So we introduce fiber_wait() which distinguishes between both values via
return value and output argument so the two can be distinguished from each
other.

Cleanup always consumes the caller's reference, including on invalid self or
cross-loop waits. The existing assertion still forbids dropping the last
reference to an unresolved future. The installed header documents this rule.

We introduce a new RUNNING fiber state to distinguish executing fibers from
non-executing fibers. We replace CANCELLED with queued resume results and
preserve interruptions across ordinary wakeups.

We add future groups with wait-all, wait-any and ignore-errors policies to
wait for multiple futures at once and migrate various existing call sites over
to them. Group size is a direct size_t query, treating NULL as an empty group.
Document all four policy combinations, including that IGNORE_ERRORS suppresses
fail-fast and parent cancellation without discarding the group's error result.

We also add defer futures and migrate bus, varlink, QMP and fiber I/O callers to
use them. Bus and QMP futures report -ENOPKG when no event loop is attached,
rather than passing NULL to sd_future_new() and returning -EINVAL.

Parent-await suppression is derived from the active wait's callback slot, so
groups use sd_fiber_await() without a separate await API or additional state.
Waits, group children and captured parents must use the same event loop.
Groups created on another loop have no parent, and a group releases its
parent reference when that fiber finishes.

Resolve SD_EVENT_DEFAULT in sd_future_new() so futures retain the actual
loop. Match group parents against that resolved loop as well; an unavailable
default returns -ENOPKG without creating an event loop implicitly.

Keep the public await entry point in fiber.c, sharing the slot-based wait with
cancellation cleanup through future-internal.h. Keep event/future integration
in event-future.c/h: priority inheritance uses the public fiber accessor, and
sd_future_new_defer() lives alongside the other event-backed constructors.
This avoids sd-event depending on sd-future private headers and removes the
redundant private defer wrapper.

Assert the non-failing slot-disabling and active-loop invariants instead of
returning early without completing callback dispatch.

Regression tests cover group error propagation and NULL-safe size queries,
interrupted waits, partial add-many rollback, groups outliving their parent,
and cancellation or timeout during nested event-loop waits, including
event-loop and descriptor cleanup. They also cover cross-loop parent isolation,
reference release on invalid cleanup calls, mixed WAIT_ANY outcomes, failed
child cancellation and policy changes on resolved groups. Additional tests
verify failed group-priority updates, slot priority during dispatch and exit,
and cancellation, timeout and spurious wakeups during bus and QMP calls. Late
replies must leave subsequent calls unaffected. Default-event tests cover
missing-loop errors, reference lifetime and parent capture by constructors.
Unattached bus and QMP clients are covered by missing-event-loop regressions.
Constructor tests verify inherited and default priorities for I/O, defer and
timer futures, plus defer argument validation and missing-default errors.

Signed-off-by: Daan De Meyer <daan@amutable.com>

src/libsystemd/forward.h

@@ -125,7 +125,8 @@ typedef struct sd_resolve_query sd_resolve_query;
 typedef struct sd_hwdb sd_hwdb;
 
 typedef struct sd_future sd_future;
+typedef struct sd_future_slot sd_future_slot;
 
-typedef int (*sd_future_func_t)(sd_future *f);
+typedef int (*sd_future_func_t)(sd_future *f, void *userdata);
 typedef int (*sd_fiber_func_t)(void *userdata);
 typedef _sd_destroy_t sd_fiber_destroy_t;

src/libsystemd/meson.build

@@ -73,6 +73,7 @@ sd_device_sources = files(
 sd_future_sources = files(
         'sd-future/fiber-io.c',
         'sd-future/fiber.c',
+        'sd-future/future-group.c',
         'sd-future/sd-future.c',
 )
 
@@ -189,6 +190,7 @@ simple_tests += files(
         'sd-future/test-fiber.c',
         'sd-future/test-fiber-io.c',
         'sd-future/test-fiber-ops.c',
+        'sd-future/test-future-group.c',
         'sd-hwdb/test-sd-hwdb.c',
         'sd-id128/test-id128.c',
         'sd-journal/test-catalog.c',

src/libsystemd/sd-bus/bus-future.c

@@ -63,9 +63,10 @@ int bus_call_future(sd_bus *bus, sd_bus_message *m, uint64_t usec, sd_future **r
         assert(bus);
         assert(m);
         assert(ret);
+        assert_return(bus->event, -ENOPKG);
 
-        _cleanup_(sd_future_unrefp) sd_future *f = NULL;
-        r = sd_future_new(&bus_future_ops, &f);
+        _cleanup_(sd_future_cancel_unrefp) sd_future *f = NULL;
+        r = sd_future_new(sd_bus_get_event(bus), &bus_future_ops, &f);
         if (r < 0)
                 return r;
 
@@ -122,7 +123,7 @@ int bus_call_suspend(
         if (r < 0)
                 return sd_bus_error_set_errno(reterr_error, r);
 
-        r = sd_fiber_suspend();
+        r = sd_fiber_await(f);
 
         /* If the future isn't resolved, the suspend was interrupted before a reply arrived (fiber
          * cancelled, fiber-wide SD_FIBER_TIMEOUT scope expired, …). There's no reply to extract,

src/libsystemd/sd-bus/bus-future.h

@@ -3,6 +3,7 @@
 
 #include "forward.h"
 
+/* The bus must have an attached event loop; otherwise return -ENOPKG. */
 int bus_call_future(sd_bus *bus, sd_bus_message *m, uint64_t usec, sd_future **ret);
 int future_get_bus_reply(sd_future *f, sd_bus_error *reterr_error, sd_bus_message **ret_reply);
 

src/libsystemd/sd-bus/bus-objects.c

@@ -373,8 +373,8 @@ DEFINE_PRIVATE_HASH_OPS_WITH_KEY_DESTRUCTOR(
                 trivial_compare_func,
                 bus_fiber_future_unref);
 
-static int bus_fiber_resolved(sd_future *f) {
-        sd_bus *bus = ASSERT_PTR(sd_future_get_userdata(f));
+static int bus_fiber_resolved(sd_future *f, void *userdata) {
+        sd_bus *bus = ASSERT_PTR(userdata);
 
         assert_se(set_remove(bus->fiber_futures, f) == f);
         sd_future_unref(f);
@@ -488,7 +488,7 @@ static int method_callbacks_run(
                                 .userdata = u,
                         };
 
-                        _cleanup_(sd_future_unrefp) sd_future *f = NULL;
+                        _cleanup_(sd_future_cancel_unrefp) sd_future *f = NULL;
                         r = sd_fiber_new(bus->event, c->member, bus_fiber_entry, d, bus_fiber_data_destroy, &f);
                         if (r < 0)
                                 return bus_maybe_reply_error(m, r, NULL);
@@ -502,8 +502,10 @@ static int method_callbacks_run(
                                 return bus_maybe_reply_error(m, r, NULL);
                         assert(r > 0);
 
-                        /* Track the future on the bus so shutdown can cancel it and wait for it. */
-                        r = sd_future_set_callback(f, bus_fiber_resolved, bus);
+                        /* Track the future on the bus so shutdown can cancel it and wait for it.
+                         * Floating slot — tied to f's lifetime; bus_fiber_resolved is what drops
+                         * the set's ref, which is the last ref and frees f (and the slot). */
+                        r = sd_future_add_callback(f, /* ret_slot= */ NULL, bus_fiber_resolved, bus);
                         if (r < 0) {
                                 /* TAKE_PTR(f) hasn't run yet, so our cleanup attribute still owns the
                                  * ref; set_remove() returns the raw pointer without firing the hash_ops

src/libsystemd/sd-bus/test-bus-fiber.c

@@ -6,6 +6,7 @@
 #include "sd-event.h"
 #include "sd-future.h"
 
+#include "bus-future.h"
 #include "bus-internal.h"
 #include "tests.h"
 #include "time-util.h"
@@ -18,6 +19,7 @@ typedef struct Context {
         int in_flight;
         int max_in_flight;
         sd_future *waiter;
+        sd_bus_message *pending_call;
 } Context;
 
 static int method_concurrent(sd_bus_message *m, void *userdata, sd_bus_error *reterr_error) {
@@ -67,6 +69,25 @@ static int method_fail_error(sd_bus_message *m, void *userdata, sd_bus_error *re
         return sd_bus_error_set(reterr_error, SD_BUS_ERROR_INVALID_ARGS, "bad arguments from fiber");
 }
 
+static int method_hold(sd_bus_message *m, void *userdata, sd_bus_error *reterr_error) {
+        Context *c = ASSERT_PTR(userdata);
+
+        ASSERT_NULL(c->pending_call);
+        c->pending_call = sd_bus_message_ref(m);
+        if (c->waiter)
+                ASSERT_OK(sd_fiber_resume(TAKE_PTR(c->waiter), 0));
+        return 1;
+}
+
+static int method_release(sd_bus_message *m, void *userdata, sd_bus_error *reterr_error) {
+        Context *c = ASSERT_PTR(userdata);
+
+        ASSERT_NOT_NULL(c->pending_call);
+        ASSERT_OK(sd_bus_reply_method_return(c->pending_call, /* types= */ NULL));
+        c->pending_call = sd_bus_message_unref(c->pending_call);
+        return sd_bus_reply_method_return(m, /* types= */ NULL);
+}
+
 static const sd_bus_vtable vtable[] = {
         SD_BUS_VTABLE_START(0),
         SD_BUS_METHOD("Concurrent", NULL, NULL, method_concurrent,
@@ -75,6 +96,8 @@ static const sd_bus_vtable vtable[] = {
                       SD_BUS_VTABLE_UNPRIVILEGED|SD_BUS_VTABLE_METHOD_FIBER),
         SD_BUS_METHOD("FailError", NULL, NULL, method_fail_error,
                       SD_BUS_VTABLE_UNPRIVILEGED|SD_BUS_VTABLE_METHOD_FIBER),
+        SD_BUS_METHOD("Hold", NULL, NULL, method_hold, SD_BUS_VTABLE_UNPRIVILEGED),
+        SD_BUS_METHOD("Release", NULL, NULL, method_release, SD_BUS_VTABLE_UNPRIVILEGED),
         SD_BUS_VTABLE_END,
 };
 
@@ -168,6 +191,17 @@ static int errors_fiber(void *userdata) {
 
         ASSERT_OK(attach_pair(s, &server, &client));
 
+        /* A detached bus must report the missing loop at the future API boundary. Reattaching
+         * it restores the regular suspending call path used below. */
+        _cleanup_(sd_bus_message_unrefp) sd_bus_message *call = NULL;
+        _cleanup_(sd_future_cancel_unrefp) sd_future *future = NULL;
+        ASSERT_OK(sd_bus_message_new_method_call(client, &call, /* destination= */ NULL,
+                                                "/test", "test.Fiber", "FailErrno"));
+        ASSERT_OK(sd_bus_detach_event(client));
+        ASSERT_ERROR(ASSERT_RETURN_EXPECTED(bus_call_future(client, call, USEC_INFINITY, &future)), ENOPKG);
+        ASSERT_NULL(future);
+        ASSERT_OK(sd_bus_attach_event(client, sd_fiber_get_event(), 0));
+
         /* A fiber handler that returns a negative errno gets turned into a matching sd_bus error
          * reply (bus_maybe_reply_error → sd_bus_reply_method_errno). */
         _cleanup_(sd_bus_error_free) sd_bus_error e1 = SD_BUS_ERROR_NULL;
@@ -204,4 +238,87 @@ TEST(fiber_method_errors) {
         ASSERT_OK(sd_future_result(f));
 }
 
+typedef struct InterruptedCall {
+        sd_bus *client;
+        int error;
+} InterruptedCall;
+
+static int interrupted_call_fiber(void *userdata) {
+        InterruptedCall *c = ASSERT_PTR(userdata);
+        _cleanup_(sd_bus_error_free) sd_bus_error error = SD_BUS_ERROR_NULL;
+        _cleanup_(sd_bus_message_unrefp) sd_bus_message *reply = NULL;
+
+        {
+                SD_FIBER_TIMEOUT(c->error == ETIME ? 0 : USEC_INFINITY);
+                ASSERT_ERROR(sd_bus_call_method(c->client, /* destination= */ NULL,
+                                               "/test", "test.Fiber", "Hold", &error, &reply,
+                                               /* types= */ NULL), c->error);
+        }
+        /* D-Bus maps both ETIME and ETIMEDOUT to its Timeout error name; the call's return value
+         * above still preserves the original interruption errno. */
+        ASSERT_EQ(sd_bus_error_get_errno(&error), c->error == ETIME ? ETIMEDOUT : c->error);
+        if (c->error == ETIME)
+                ASSERT_TRUE(sd_bus_error_has_name(&error, SD_BUS_ERROR_TIMEOUT));
+        ASSERT_NULL(reply);
+        sd_bus_error_free(&error);
+
+        /* Release sends the abandoned call's reply before its own. The old reply must neither
+         * access a freed future nor wake this new wait, and no interruption may leak into it. */
+        ASSERT_OK_POSITIVE(sd_bus_call_method(c->client, /* destination= */ NULL,
+                                            "/test", "test.Fiber", "Release", &error, &reply,
+                                            /* types= */ NULL));
+        ASSERT_NOT_NULL(reply);
+        ASSERT_FALSE(sd_bus_error_is_set(&error));
+        ASSERT_OK_ZERO(sd_fiber_yield());
+        return 0;
+}
+
+static int interrupted_calls_fiber(void *userdata) {
+        Setup *s = ASSERT_PTR(userdata);
+        _cleanup_(sd_bus_flush_close_unrefp) sd_bus *server = NULL, *client = NULL;
+        int error;
+
+        ASSERT_OK(attach_pair(s, &server, &client));
+
+        FOREACH_ARGUMENT(error, ECANCELED, ETIME, EBUSY) {
+                _cleanup_(sd_future_cancel_wait_unrefp) sd_future *caller = NULL;
+                InterruptedCall c = { .client = client, .error = error };
+
+                s->c->waiter = error == ETIME ? NULL : sd_fiber_get_current();
+                ASSERT_OK(sd_fiber_new(sd_fiber_get_event(), "interrupted-call", interrupted_call_fiber,
+                                       &c, /* destroy= */ NULL, &caller));
+
+                if (error != ETIME) {
+                        /* Interrupt only once the server has the call in hand and withheld its reply. */
+                        ASSERT_OK_ZERO(sd_fiber_suspend());
+                        ASSERT_NOT_NULL(s->c->pending_call);
+                        if (error == ECANCELED)
+                                ASSERT_OK(sd_future_cancel(caller));
+                        else
+                                ASSERT_OK(sd_fiber_resume(caller, 42));
+                }
+
+                ASSERT_OK_ZERO(sd_fiber_await(caller));
+                ASSERT_NULL(s->c->pending_call);
+                ASSERT_NULL(s->c->waiter);
+        }
+
+        return 0;
+}
+
+TEST(fiber_method_interrupted) {
+        _cleanup_(sd_event_unrefp) sd_event *e = NULL;
+        _cleanup_(sd_future_unrefp) sd_future *f = NULL;
+        Context c = {};
+        Setup s = { .c = &c };
+
+        ASSERT_OK_ERRNO(socketpair(AF_UNIX, SOCK_STREAM|SOCK_CLOEXEC, 0, s.fds));
+        ASSERT_OK(sd_event_new(&e));
+        ASSERT_OK(sd_event_set_exit_on_idle(e, true));
+        ASSERT_OK(sd_fiber_new(e, "interrupted-calls", interrupted_calls_fiber, &s,
+                               /* destroy= */ NULL, &f));
+        ASSERT_OK(sd_event_loop(e));
+        ASSERT_OK_ZERO(sd_future_result(f));
+}
+
 DEFINE_TEST_MAIN(LOG_DEBUG);

src/libsystemd/sd-event/event-future.c

@@ -9,6 +9,22 @@
 #include "event-util.h"
 #include "fd-util.h"
 
+int event_source_inherit_fiber_priority(sd_event_source *s) {
+        int64_t priority;
+        int r;
+
+        assert(s);
+
+        if (!sd_fiber_is_running())
+                return 0;
+
+        r = sd_fiber_get_priority(&priority);
+        if (r < 0)
+                return r;
+
+        return sd_event_source_set_priority(s, priority);
+}
+
 typedef struct IoFuture {
         sd_event_source *source;
 } IoFuture;
@@ -84,8 +100,8 @@ int future_new_io(sd_event *e, int fd, uint32_t events, sd_future **ret) {
         if (IN_SET(sd_event_get_state(e), SD_EVENT_EXITING, SD_EVENT_FINISHED))
                 return -ECANCELED;
 
-        _cleanup_(sd_future_unrefp) sd_future *f = NULL;
-        r = sd_future_new(&io_future_ops, &f);
+        _cleanup_(sd_future_cancel_unrefp) sd_future *f = NULL;
+        r = sd_future_new(e, &io_future_ops, &f);
         if (r < 0)
                 return r;
 
@@ -110,17 +126,101 @@ int future_new_io(sd_event *e, int fd, uint32_t events, sd_future **ret) {
         if (r < 0)
                 return r;
 
-        if (sd_fiber_is_running()) {
-                int64_t priority;
+        r = event_source_inherit_fiber_priority(iof->source);
+        if (r < 0)
+                return r;
 
-                r = sd_fiber_get_priority(&priority);
-                if (r < 0)
-                        return r;
+        *ret = TAKE_PTR(f);
+        return 0;
+}
 
-                r = sd_event_source_set_priority(iof->source, priority);
-                if (r < 0)
-                        return r;
-        }
+int future_group_add_io(sd_future *group, int fd, uint32_t events) {
+        _cleanup_(sd_future_cancel_unrefp) sd_future *io = NULL;
+        int r;
+
+        assert(group);
+
+        r = future_new_io(sd_future_get_event(group), fd, events, &io);
+        if (r < 0)
+                return r;
+
+        r = sd_future_group_add(group, io);
+        if (r < 0)
+                return r;
+
+        /* The group owns a reference now: release ours without cancelling the child. */
+        io = sd_future_unref(io);
+        return 0;
+}
+
+typedef struct DeferFuture {
+        sd_event_source *source;
+        int result;
+} DeferFuture;
+
+static void* defer_future_alloc(void) {
+        return new0(DeferFuture, 1);
+}
+
+static void defer_future_free(sd_future *f) {
+        DeferFuture *df = ASSERT_PTR(sd_future_get_private(ASSERT_PTR(f)));
+
+        sd_event_source_unref(df->source);
+        free(df);
+}
+
+static int defer_future_cancel(sd_future *f) {
+        DeferFuture *df = ASSERT_PTR(sd_future_get_private(ASSERT_PTR(f)));
+        int r;
+
+        r = sd_event_source_set_enabled(df->source, SD_EVENT_OFF);
+        RET_GATHER(r, sd_future_resolve(f, -ECANCELED));
+        return r;
+}
+
+static int defer_future_set_priority(sd_future *f, int64_t priority) {
+        DeferFuture *df = ASSERT_PTR(sd_future_get_private(ASSERT_PTR(f)));
+        return sd_event_source_set_priority(df->source, priority);
+}
+
+static const sd_future_ops defer_future_ops = {
+        .size = sizeof(sd_future_ops),
+        .alloc = defer_future_alloc,
+        .free = defer_future_free,
+        .cancel = defer_future_cancel,
+        .set_priority = defer_future_set_priority,
+};
+
+static int defer_handler(sd_event_source *s, void *userdata) {
+        sd_future *f = ASSERT_PTR(userdata);
+        DeferFuture *df = ASSERT_PTR(sd_future_get_private(f));
+        return sd_future_resolve(f, df->result);
+}
+
+int sd_future_new_defer(sd_event *e, int result, sd_future **ret) {
+        int r;
+
+        assert_return(e, -EINVAL);
+        assert_return(ret, -EINVAL);
+
+        if (IN_SET(sd_event_get_state(e), SD_EVENT_EXITING, SD_EVENT_FINISHED))
+                return -ECANCELED;
+
+        _cleanup_(sd_future_cancel_unrefp) sd_future *f = NULL;
+        r = sd_future_new(e, &defer_future_ops, &f);
+        if (r < 0)
+                return r;
+
+        DeferFuture *df = sd_future_get_private(f);
+        df->result = result;
+
+        r = sd_event_add_defer(e, &df->source, defer_handler, f);
+        if (r < 0)
+                return r;
+
+        r = event_source_inherit_fiber_priority(df->source);
+        if (r < 0)
+                return r;
 
         *ret = TAKE_PTR(f);
         return 0;
@@ -200,8 +300,8 @@ static int future_new_time_internal(
         if (IN_SET(sd_event_get_state(e), SD_EVENT_EXITING, SD_EVENT_FINISHED))
                 return -ECANCELED;
 
-        _cleanup_(sd_future_unrefp) sd_future *f = NULL;
-        r = sd_future_new(&time_future_ops, &f);
+        _cleanup_(sd_future_cancel_unrefp) sd_future *f = NULL;
+        r = sd_future_new(e, &time_future_ops, &f);
         if (r < 0)
                 return r;
 
@@ -212,17 +312,9 @@ static int future_new_time_internal(
         if (r < 0)
                 return r;
 
-        if (sd_fiber_is_running()) {
-                int64_t priority;
-
-                r = sd_fiber_get_priority(&priority);
-                if (r < 0)
-                        return r;
-
-                r = sd_event_source_set_priority(tf->source, priority);
-                if (r < 0)
-                        return r;
-        }
+        r = event_source_inherit_fiber_priority(tf->source);
+        if (r < 0)
+                return r;
 
         *ret = TAKE_PTR(f);
         return 0;
@@ -236,6 +328,25 @@ int future_new_time_relative(sd_event *e, clockid_t clock, uint64_t usec, uint64
         return future_new_time_internal(sd_event_add_time_relative, e, clock, usec, accuracy, result, ret);
 }
 
+int future_group_add_time_relative(sd_future *group, clockid_t clock, uint64_t usec, uint64_t accuracy, int result) {
+        _cleanup_(sd_future_cancel_unrefp) sd_future *timer = NULL;
+        int r;
+
+        assert(group);
+
+        r = future_new_time_relative(sd_future_get_event(group), clock, usec, accuracy, result, &timer);
+        if (r < 0)
+                return r;
+
+        r = sd_future_group_add(group, timer);
+        if (r < 0)
+                return r;
+
+        /* The group owns a reference now: release ours without cancelling the child. */
+        timer = sd_future_unref(timer);
+        return 0;
+}
+
 int event_run_suspend(sd_event *e, uint64_t timeout) {
         sd_event *outer = sd_fiber_get_event();
         int r;
@@ -271,25 +382,32 @@ int event_run_suspend(sd_event *e, uint64_t timeout) {
         if (fd < 0)
                 return fd;
 
-        _cleanup_(sd_future_cancel_wait_unrefp) sd_future *io = NULL;
-        r = future_new_io(outer, fd, EPOLLIN, &io);
+        /* Wait for the inner-loop fd to become readable OR (optionally) the timeout to fire. */
+        _cleanup_(sd_future_cancel_wait_unrefp) sd_future *group = NULL;
+        r = sd_future_group_new(outer, &group);
+        if (r < 0)
+                return r;
+
+        r = sd_future_group_set_policy(group, SD_FUTURE_GROUP_WAIT_ANY);
+        if (r < 0)
+                return r;
+
+        r = future_group_add_io(group, fd, EPOLLIN);
         if (r < 0)
                 return r;
 
-        _cleanup_(sd_future_cancel_wait_unrefp) sd_future *timer = NULL;
         if (timeout != USEC_INFINITY) {
-                r = future_new_time_relative(
-                                outer,
+                r = future_group_add_time_relative(
+                                group,
                                 CLOCK_MONOTONIC,
                                 timeout,
                                 /* accuracy= */ 1,
-                                /* result= */ 0,
-                                &timer);
+                                /* result= */ 0);
                 if (r < 0)
                         return r;
         }
 
-        r = sd_fiber_suspend();
+        r = sd_fiber_await(group);
         if (r < 0)
                 return r;
 

src/libsystemd/sd-event/event-future.h

@@ -3,8 +3,15 @@
 
 #include "forward.h"
 
+/* Event/future integration, implemented using only the public sd-future API. */
+
 int future_new_io(sd_event *e, int fd, uint32_t events, sd_future **ret);
 int future_new_time(sd_event *e, clockid_t clock, uint64_t usec, uint64_t accuracy, int result, sd_future **ret);
 int future_new_time_relative(sd_event *e, clockid_t clock, uint64_t usec, uint64_t accuracy, int result, sd_future **ret);
 
+int future_group_add_io(sd_future *group, int fd, uint32_t events);
+int future_group_add_time_relative(sd_future *group, clockid_t clock, uint64_t usec, uint64_t accuracy, int result);
+
+int event_source_inherit_fiber_priority(sd_event_source *s);
+
 int event_run_suspend(sd_event *e, uint64_t timeout);

src/libsystemd/sd-event/event-util.h

@@ -10,6 +10,8 @@
 
 extern const struct hash_ops event_source_hash_ops;
 
+sd_event* event_resolve(sd_event *e);
+
 int event_reset_time(
                 sd_event *e,
                 sd_event_source **s,

src/libsystemd/sd-event/sd-event.c

@@ -192,7 +192,7 @@ static thread_local sd_event *default_event = NULL;
 static void source_disconnect(sd_event_source *s);
 static void event_gc_inode_data(sd_event *e, InodeData *d);
 
-static sd_event* event_resolve(sd_event *e) {
+sd_event* event_resolve(sd_event *e) {
         return e == SD_EVENT_DEFAULT ? default_event : e;
 }
 

src/libsystemd/sd-event/test-event-future.c

@@ -6,10 +6,170 @@
 #include "sd-event.h"
 #include "sd-future.h"
 
+#include "event-future.h"
 #include "fd-util.h"
 #include "tests.h"
 #include "time-util.h"
 
+typedef enum EventFutureType {
+        EVENT_FUTURE_IO,
+        EVENT_FUTURE_DEFER,
+        EVENT_FUTURE_TIME,
+        EVENT_FUTURE_TIME_RELATIVE,
+        _EVENT_FUTURE_TYPE_MAX,
+        _EVENT_FUTURE_TYPE_INVALID = -EINVAL,
+} EventFutureType;
+
+static int new_event_future(EventFutureType type, sd_event *e, int fd, sd_future **ret) {
+        switch (type) {
+        case EVENT_FUTURE_IO:
+                return future_new_io(e, fd, EPOLLIN, ret);
+        case EVENT_FUTURE_DEFER:
+                return sd_future_new_defer(e, 0, ret);
+        case EVENT_FUTURE_TIME:
+                return future_new_time(e, CLOCK_MONOTONIC, 0, 1, 0, ret);
+        case EVENT_FUTURE_TIME_RELATIVE:
+                return future_new_time_relative(e, CLOCK_MONOTONIC, 0, 1, 0, ret);
+        default:
+                assert_not_reached();
+        }
+}
+
+typedef struct EventFuturePriorityData {
+        EventFutureType type;
+        int fd;
+        sd_future **ret;
+} EventFuturePriorityData;
+
+static int event_future_priority_fiber(void *userdata) {
+        EventFuturePriorityData *d = ASSERT_PTR(userdata);
+
+        return new_event_future(d->type, sd_fiber_get_event(), d->fd, d->ret);
+}
+
+TEST(event_future_priority) {
+        int64_t priority, probe_priority;
+
+        for (EventFutureType type = 0; type < _EVENT_FUTURE_TYPE_MAX; type++)
+                FOREACH_ARGUMENT(priority, -10, 0, 10)
+                        FOREACH_ARGUMENT(probe_priority, -5, 5) {
+                                _cleanup_(sd_event_unrefp) sd_event *e = NULL;
+                                _cleanup_close_pair_ int fds[2] = EBADF_PAIR;
+                                _cleanup_(sd_future_cancel_unrefp) sd_future *f = NULL, *probe = NULL;
+
+                                ASSERT_OK(sd_event_new(&e));
+                                if (type == EVENT_FUTURE_IO) {
+                                        ASSERT_OK_ERRNO(pipe2(fds, O_CLOEXEC | O_NONBLOCK));
+                                        ASSERT_OK_EQ_ERRNO(write(fds[1], "X", 1), 1);
+                                }
+
+                                if (priority != 0) {
+                                        _cleanup_(sd_future_unrefp) sd_future *registrar = NULL;
+                                        EventFuturePriorityData d = { .type = type, .fd = fds[0], .ret = &f };
+
+                                        ASSERT_OK(sd_fiber_new(e, "new-event-future", event_future_priority_fiber,
+                                                              &d, /* destroy= */ NULL, &registrar));
+                                        ASSERT_OK(sd_future_set_priority(registrar, priority));
+                                        ASSERT_OK_POSITIVE(sd_event_run(e, 0));
+                                        ASSERT_OK_ZERO(sd_future_result(registrar));
+                                } else
+                                        /* Construction outside a fiber must keep the normal priority. */
+                                        ASSERT_OK(new_event_future(type, e, fds[0], &f));
+
+                                ASSERT_EQ(sd_future_state(f), SD_FUTURE_PENDING);
+                                ASSERT_OK(new_event_future(type, e, fds[0], &probe));
+                                ASSERT_OK(sd_future_set_priority(probe, probe_priority));
+
+                                /* Both sources are ready. Inspect resolution directly, without callback
+                                 * slots whose own priority inheritance could hide a constructor bug. */
+                                ASSERT_OK_POSITIVE(sd_event_run(e, 0));
+                                ASSERT_EQ(sd_future_state(f),
+                                          priority < probe_priority ? SD_FUTURE_RESOLVED : SD_FUTURE_PENDING);
+                                ASSERT_EQ(sd_future_state(probe),
+                                          priority < probe_priority ? SD_FUTURE_PENDING : SD_FUTURE_RESOLVED);
+                        }
+}
+
+TEST(event_source_inherit_fiber_priority_without_fiber) {
+        _cleanup_(sd_event_unrefp) sd_event *e = NULL;
+        _cleanup_(sd_event_source_unrefp) sd_event_source *s = NULL;
+        int64_t priority;
+
+        ASSERT_OK(sd_event_new(&e));
+        ASSERT_OK(sd_event_add_defer(e, &s, /* callback= */ NULL, /* userdata= */ NULL));
+        ASSERT_OK(sd_event_source_set_priority(s, 42));
+        ASSERT_OK(event_source_inherit_fiber_priority(s));
+        ASSERT_OK(sd_event_source_get_priority(s, &priority));
+        ASSERT_EQ(priority, 42);
+}
+
+TEST(future_new_defer_invalid) {
+        _cleanup_(sd_event_unrefp) sd_event *e = NULL;
+        _cleanup_(sd_future_cancel_unrefp) sd_future *f = NULL;
+
+        ASSERT_ERROR(ASSERT_RETURN_EXPECTED(sd_future_new_defer(/* e= */ NULL, 0, &f)), EINVAL);
+        ASSERT_OK_ZERO(sd_event_default(/* ret= */ NULL));
+        ASSERT_ERROR(ASSERT_RETURN_EXPECTED(sd_future_new_defer(SD_EVENT_DEFAULT, 0, &f)), ENOPKG);
+        ASSERT_OK_ZERO(sd_event_default(/* ret= */ NULL));
+
+        ASSERT_OK(sd_event_new(&e));
+        ASSERT_ERROR(ASSERT_RETURN_EXPECTED(sd_future_new_defer(e, 0, /* ret= */ NULL)), EINVAL);
+        ASSERT_NULL(f);
+}
+
+TEST(future_group_add_event) {
+        bool timer;
+
+        FOREACH_ARGUMENT(timer, false, true) {
+                _cleanup_(sd_event_unrefp) sd_event *e = NULL;
+                _cleanup_(sd_future_unrefp) sd_future *group = NULL;
+                _cleanup_close_pair_ int pipefd[2] = EBADF_PAIR;
+
+                ASSERT_OK(sd_event_new(&e));
+                ASSERT_OK(sd_future_group_new(e, &group));
+                ASSERT_OK(sd_future_group_set_policy(group, SD_FUTURE_GROUP_WAIT_ANY));
+
+                if (timer)
+                        ASSERT_OK(future_group_add_time_relative(group, CLOCK_MONOTONIC, 0, 1, 42));
+                else {
+                        ASSERT_OK_ERRNO(pipe2(pipefd, O_CLOEXEC | O_NONBLOCK));
+                        ASSERT_OK(future_group_add_io(group, pipefd[0], EPOLLIN));
+                        ASSERT_OK_ZERO(sd_event_run(e, 0));
+                        ASSERT_OK_EQ_ERRNO(write(pipefd[1], "X", 1), 1);
+                }
+
+                ASSERT_EQ(sd_future_group_size(group), 1U);
+
+                /* The helper must leave the child alive and uncancelled, with the group owning its
+                 * only reference, until its event fires. */
+                ASSERT_OK(sd_event_set_exit_on_idle(e, true));
+                ASSERT_OK(sd_event_loop(e));
+                ASSERT_EQ(sd_future_result(group), timer ? 42 : (int) EPOLLIN);
+        }
+}
+
+TEST(future_group_add_event_failure) {
+        _cleanup_(sd_event_unrefp) sd_event *e = NULL;
+        _cleanup_(sd_future_unrefp) sd_future *group = NULL;
+        _cleanup_close_pair_ int pipefd[2] = EBADF_PAIR;
+        char c;
+
+        ASSERT_OK(sd_event_new(&e));
+        ASSERT_OK(sd_future_group_new(e, &group));
+        ASSERT_OK(sd_future_cancel(group));
+        ASSERT_OK_ERRNO(pipe2(pipefd, O_CLOEXEC | O_NONBLOCK));
+        ASSERT_OK_EQ_ERRNO(write(pipefd[1], "X", 1), 1);
+
+        ASSERT_ERROR(ASSERT_RETURN_EXPECTED(future_group_add_io(group, pipefd[0], EPOLLIN)), ESTALE);
+        ASSERT_ERROR(ASSERT_RETURN_EXPECTED(future_group_add_time_relative(group, CLOCK_MONOTONIC, 0, 1, 42)), ESTALE);
+
+        /* Rejected children must leave no enabled sources behind, and releasing the IO child's
+         * duplicate fd must not close the caller's fd. */
+        ASSERT_OK_ZERO(sd_event_run(e, 0));
+        ASSERT_OK_EQ_ERRNO(read(pipefd[0], &c, 1), 1);
+        ASSERT_EQ(c, 'X');
+}
+
 static int timer_callback(sd_event_source *s, uint64_t usec, void *userdata) {
         int *count = ASSERT_PTR(userdata);
         int r;
@@ -355,4 +515,67 @@ TEST(sd_event_run_timer) {
         ASSERT_OK_ZERO(sd_future_result(f));
 }
 
+typedef struct EventRunInterruptedState {
+        sd_event *inner;
+        bool timeout;
+        bool finite;
+} EventRunInterruptedState;
+
+static int event_run_interrupted_fiber(void *userdata) {
+        EventRunInterruptedState *s = ASSERT_PTR(userdata);
+
+        SD_FIBER_TIMEOUT(s->timeout ? 5 * USEC_PER_MSEC : USEC_INFINITY);
+        return sd_event_run(s->inner, s->finite ? USEC_PER_MINUTE : USEC_INFINITY);
+}
+
+TEST(sd_event_run_interrupted) {
+        bool timeout, finite;
+
+        FOREACH_ARGUMENT(timeout, false, true)
+                FOREACH_ARGUMENT(finite, false, true) {
+                        _cleanup_(sd_event_unrefp) sd_event *outer = NULL, *inner = NULL;
+                        _cleanup_(sd_event_source_unrefp) sd_event_source *source = NULL;
+                        _cleanup_(sd_future_unrefp) sd_future *f = NULL;
+                        _cleanup_close_pair_ int pipefd[2] = EBADF_PAIR;
+                        int count = 0;
+
+                        ASSERT_OK(sd_event_new(&outer));
+                        ASSERT_OK(sd_event_new(&inner));
+                        ASSERT_OK_ERRNO(pipe2(pipefd, O_CLOEXEC | O_NONBLOCK));
+                        ASSERT_OK(sd_event_add_io(inner, &source, pipefd[0], EPOLLIN, io_callback, &count));
+
+                        EventRunInterruptedState s = { .inner = inner, .timeout = timeout, .finite = finite };
+                        ASSERT_OK(sd_fiber_new(outer, "event-interrupted", event_run_interrupted_fiber, &s,
+                                               /* destroy= */ NULL, &f));
+                        ASSERT_OK_POSITIVE(sd_event_run(outer, 0));
+                        ASSERT_EQ(sd_future_state(f), SD_FUTURE_PENDING);
+                        if (!timeout)
+                                ASSERT_OK(sd_future_cancel(f));
+
+                        while (sd_future_state(f) == SD_FUTURE_PENDING)
+                                ASSERT_OK_POSITIVE(sd_event_run(outer, USEC_INFINITY));
+                        ASSERT_EQ(sd_future_result(f), timeout ? -ETIME : -ECANCELED);
+                        ASSERT_EQ(count, 0);
+
+                        /* Cleanup must remove the outer loop's IO watcher and optional timer. Making
+                         * the inner fd readable must not dispatch anything on the outer loop, while
+                         * the inner loop itself remains usable after the interrupted run. */
+                        ASSERT_OK_EQ_ERRNO(write(pipefd[1], "X", 1), 1);
+                        ASSERT_OK_ZERO(sd_event_run(outer, 0));
+                        ASSERT_OK_POSITIVE(sd_event_run(inner, 0));
+                        ASSERT_EQ(count, 1);
+
+                        int inner_fd = ASSERT_OK(sd_event_get_fd(inner));
+                        int outer_fd = ASSERT_OK(sd_event_get_fd(outer));
+                        f = sd_future_unref(f);
+                        source = sd_event_source_unref(source);
+                        inner = sd_event_unref(inner);
+                        outer = sd_event_unref(outer);
+
+                        /* A leaked future or slot would keep its event loop and epoll fd alive. */
+                        ASSERT_ERROR_ERRNO(fcntl(inner_fd, F_GETFD), EBADF);
+                        ASSERT_ERROR_ERRNO(fcntl(outer_fd, F_GETFD), EBADF);
+                }
+}
+
 DEFINE_TEST_MAIN(LOG_DEBUG);

src/libsystemd/sd-future/fiber-io.c

@@ -10,7 +10,6 @@
 #include "sd-event.h"
 #include "sd-future.h"
 
-#include "alloc-util.h"
 #include "errno-util.h"
 #include "event-future.h"
 #include "fd-util.h"
@@ -51,7 +50,7 @@ static ssize_t fiber_io_operation(
         if (r < 0)
                 return r;
 
-        r = sd_fiber_suspend();
+        r = sd_fiber_await(io);
         if (r < 0)
                 return r;
 
@@ -230,7 +229,7 @@ int sd_fiber_connect(int sockfd, const struct sockaddr *addr, socklen_t addrlen)
 
         /* future_new_io resolves with the revents mask on success; translate any positive value
          * (e.g. POLLOUT) back to the connect(2) success status. */
-        r = sd_fiber_suspend();
+        r = sd_fiber_await(io);
         return r > 0 ? 0 : r;
 }
 
@@ -395,17 +394,18 @@ int sd_fiber_ppoll(struct pollfd *fds, size_t n_fds, const struct timespec *time
         if (zero_timeout || r != 0) /* Either error or some fds are ready */
                 return r;
 
-        sd_future **futures = NULL;
-        CLEANUP_ARRAY(futures, n_fds, sd_future_cancel_wait_unref_array);
+        /* Use a WAIT_ANY group: the first child (an fd readiness or the timer) to settle resolves
+         * the group, which cancels its siblings on the spot. The user-supplied event mask is in
+         * poll() bits (struct pollfd), so translate to epoll bits. */
+        _cleanup_(sd_future_cancel_wait_unrefp) sd_future *group = NULL;
+        r = sd_future_group_new(e, &group);
+        if (r < 0)
+                return r;
 
-        futures = new0(sd_future*, n_fds);
-        if (!futures)
-                return -ENOMEM;
+        r = sd_future_group_set_policy(group, SD_FUTURE_GROUP_WAIT_ANY);
+        if (r < 0)
+                return r;
 
-        /* Set up I/O event sources for all valid fds. POLL* and EPOLL* share their bit values (see
-         * EPOLL_POLL_COMMON_MASK in io-util.h), so we can pass the user-supplied event mask through
-         * to either backend without translation. */
-        size_t n_io_futures = 0;
         for (size_t i = 0; i < n_fds; i++) {
                 if (fds[i].fd < 0)
                         continue;
@@ -414,11 +414,9 @@ int sd_fiber_ppoll(struct pollfd *fds, size_t n_fds, const struct timespec *time
                 if (events == 0)
                         continue;
 
-                r = future_new_io(e, fds[i].fd, events, &futures[i]);
+                r = future_group_add_io(group, fds[i].fd, events);
                 if (r < 0)
                         return r;
-
-                n_io_futures++;
         }
 
         /* A timeout that overflows usec_t saturates to USEC_INFINITY in timespec_load(); treat that
@@ -430,23 +428,21 @@ int sd_fiber_ppoll(struct pollfd *fds, size_t n_fds, const struct timespec *time
         /* If every fd was skipped (negative or empty event mask) and we'd have no timer, there's
          * nothing that could ever wake the fiber up — same situation as n_fds == 0 && !timeout,
          * just not detectable upfront. Refuse rather than suspend forever. */
-        if (n_io_futures == 0 && usec == USEC_INFINITY)
+        if (sd_future_group_size(group) == 0 && usec == USEC_INFINITY)
                 return -EINVAL;
 
-        _cleanup_(sd_future_cancel_wait_unrefp) sd_future *timer = NULL;
         if (usec != USEC_INFINITY) {
-                r = future_new_time_relative(
-                                e,
+                r = future_group_add_time_relative(
+                                group,
                                 CLOCK_MONOTONIC,
                                 usec,
                                 /* accuracy= */ 1,
-                                /* result= */ 0,
-                                &timer);
+                                /* result= */ 0);
                 if (r < 0)
                         return r;
         }
 
-        r = sd_fiber_suspend();
+        r = sd_fiber_await(group);
         if (r < 0 && r != -ETIME)
                 return r;
 
@@ -457,14 +453,10 @@ int sd_fiber_ppoll(struct pollfd *fds, size_t n_fds, const struct timespec *time
         if (n != 0)
                 return n;
 
-        /* No fds ready: distinguish our own timer from an external -ETIME. */
-        if (timer && sd_future_state(timer) == SD_FUTURE_RESOLVED)
-                return 0;
-
-        /* An IO future resolved with a revents mask (r > 0) but the readiness was already consumed
-         * by the time we swept — report 0 rather than leaking the bitmask as a (bogus) ppoll fd
-         * count to the caller. */
-        if (r > 0)
+        /* No fds ready. The group's result is the winning child's result: 0 means the timer
+         * (created with result=0) fired; r > 0 means an IO future fired (revents mask) but the
+         * readiness was already drained when we swept. Both map to "0 fds ready". */
+        if (r >= 0)
                 return 0;
 
         return r;

src/libsystemd/sd-future/fiber.c

@@ -39,13 +39,13 @@
  * synonym there. */
 _noreturn_ extern void siglongjmp_unchecked(sigjmp_buf env, int val) __asm__("siglongjmp");
 
-static thread_local Fiber *current_fiber = NULL;
+static thread_local sd_future *current_fiber = NULL;
 
 typedef enum FiberState {
         FIBER_STATE_INITIAL,
         FIBER_STATE_READY,
+        FIBER_STATE_RUNNING,
         FIBER_STATE_SUSPENDED,
-        FIBER_STATE_CANCELLED,
         FIBER_STATE_COMPLETED,
         _FIBER_STATE_MAX,
         _FIBER_STATE_INVALID = -EINVAL,
@@ -64,10 +64,11 @@ typedef struct Fiber {
 
         FiberState state;
         int result;                     /* Either resume error code or final return value */
+        bool result_pending;            /* sd_fiber_resume() stashed a value that fiber_swap() hasn't consumed yet */
 
         sd_future *floating;            /* Self-ref held while the fiber is floating; dropped on resolve. */
+        sd_future *awaiting;            /* Target of the wait the fiber is suspended in, if any. */
 
-        sd_event *event;
         sd_event_source *defer_event_source;
         sd_event_source *exit_event_source;
 
@@ -89,11 +90,15 @@ typedef struct Fiber {
 #endif
 } Fiber;
 
-static Fiber* fiber_get_current(void) {
-        return current_fiber;
+static Fiber* fiber_get(sd_future *f) {
+        return ASSERT_PTR(sd_future_get_private(ASSERT_PTR(f)));
 }
 
-static void fiber_set_current(Fiber *f) {
+static Fiber* fiber_get_maybe(sd_future *f) {
+        return f ? fiber_get(f) : NULL;
+}
+
+static void fiber_set_current(sd_future *f) {
         current_fiber = f;
 }
 
@@ -167,7 +172,7 @@ static inline void finish_switch_stack(void *fake_stack_save) {
 
 /* Refresh f->resume_stack from whoever is currently the running fiber, so the next siglongjmp() out
  * of f (in the trampoline or fiber_swap()) can hand the right destination stack to ASAN. Must be
- * called before fiber_set_current(f) — relies on fiber_get_current() returning the caller. */
+ * called before fiber_set_current(f) — relies on sd_fiber_get_current() returning the caller. */
 static void fiber_set_resume_stack(Fiber *f, Fiber *resume) {
         assert(f);
 
@@ -178,70 +183,70 @@ static void fiber_set_resume_stack(Fiber *f, Fiber *resume) {
 }
 
 _noreturn_ static void fiber_entry_point(void) {
-        Fiber *f = ASSERT_PTR(fiber_get_current());
+        Fiber *fiber = fiber_get(sd_fiber_get_current());
         void *fake_stack_save = NULL;
 
-        assert(f->func);
-        assert(IN_SET(f->state, FIBER_STATE_INITIAL, FIBER_STATE_READY, FIBER_STATE_CANCELLED));
+        assert(fiber->func);
+        assert(fiber->state == FIBER_STATE_INITIAL);
 
         finish_switch_stack(NULL);
 
         /* Capture our resumable point on the fiber's stack, then bounce back to whoever last set
-         * f->resume_context. On bootstrap that's fiber_bootstrap(); on every subsequent yield it's
+         * fiber->resume_context. On bootstrap that's fiber_init(); on every subsequent yield it's
          * the most recent fiber_run(). sigsetjmp(buf, 0) skips the signal-mask save: switching is
          * thread-shared with respect to signal masks. */
-        if (sigsetjmp(f->context, /* savemask= */ 0) == 0) {
-                start_switch_stack(&fake_stack_save, &f->resume_stack);
-                siglongjmp_unchecked(f->resume_context, /* val= */ 1);
+        if (sigsetjmp(fiber->context, /* savemask= */ 0) == 0) {
+                start_switch_stack(&fake_stack_save, &fiber->resume_stack);
+                siglongjmp_unchecked(fiber->resume_context, /* val= */ 1);
         }
 
-        /* Re-entered for real via fiber_run()'s siglongjmp(f->context). */
+        /* Re-entered for real via fiber_run()'s siglongjmp(fiber->context). */
         finish_switch_stack(fake_stack_save);
+        assert(fiber->state == FIBER_STATE_RUNNING);
 
         /* Block scope so the cleanups attached to LOG_SET_PREFIX / LOG_CONTEXT_PUSH_KEY_VALUE fire
          * before the siglongjmp below — siglongjmp skips _cleanup_ attributes, so we have to make
          * sure the scope ends via a normal control-flow path first. */
         {
-                LOG_SET_PREFIX(f->name);
-                LOG_CONTEXT_PUSH_KEY_VALUE("FIBER=", f->name);
+                LOG_SET_PREFIX(fiber->name);
+                LOG_CONTEXT_PUSH_KEY_VALUE("FIBER=", fiber->name);
 
-                f->result = f->state == FIBER_STATE_CANCELLED ? -ECANCELED : f->func(f->userdata);
-                f->state = FIBER_STATE_COMPLETED;
+                fiber->result = fiber->func(fiber->userdata);
+                fiber->state = FIBER_STATE_COMPLETED;
         }
 
         /* Pass NULL fake_stack_save to discard the fiber's fake stack since the fiber is done. */
-        start_switch_stack(NULL, &f->resume_stack);
+        start_switch_stack(NULL, &fiber->resume_stack);
 
         /* Bounce back to whichever fiber_run() call most recently entered us. resume_context is
          * per-fiber so nested fiber_run() — e.g. a bus method dispatched as a fiber handler while
          * sd_event_loop() itself runs in a fiber — is safe. */
-        siglongjmp_unchecked(f->resume_context, 1);
+        siglongjmp_unchecked(fiber->resume_context, 1);
         assert_not_reached();
 }
 
-static int fiber_init(Fiber *f) {
+static int fiber_init(sd_future *f) {
+        Fiber *fiber = fiber_get(f);
         ucontext_t old_uc, uc;
         void *fake_stack_save = NULL;
 
-        assert(f);
-
         if (getcontext(&uc) < 0)
                 return -errno;
 
-        struct iovec fiber_stack = fiber_stack_usable(&f->stack);
+        struct iovec fiber_stack = fiber_stack_usable(&fiber->stack);
 
         uc.uc_link = NULL;              /* Unused: trampoline siglongjmps out instead of returning. */
         uc.uc_stack.ss_sp = fiber_stack.iov_base;
         uc.uc_stack.ss_size = fiber_stack.iov_len;
         uc.uc_stack.ss_flags = 0;
 
-        Fiber *prev = fiber_get_current();
+        sd_future *prev = sd_fiber_get_current();
         fiber_set_current(f);
 
         makecontext(&uc, fiber_entry_point, /* argc= */ 0);
 
-        fiber_set_resume_stack(f, prev);
-        if (sigsetjmp(f->resume_context, /* savemask= */ 0) == 0) {
+        fiber_set_resume_stack(fiber, fiber_get_maybe(prev));
+        if (sigsetjmp(fiber->resume_context, /* savemask= */ 0) == 0) {
                 start_switch_stack(&fake_stack_save, &fiber_stack);
                 if (swapcontext(&old_uc, &uc) < 0) {
                         finish_switch_stack(fake_stack_save);
@@ -270,20 +275,23 @@ static void reset_current_fiber(void) {
         /* Restore the caller's log state stashed in the running fiber (if any) before clearing
          * current_fiber. Without this, the child of a fork() that happened mid-fiber would inherit the
          * fiber's log prefix / context list in its thread-locals even though no fiber is running. */
-        Fiber *f = fiber_get_current();
+        sd_future *f = sd_fiber_get_current();
         if (f) {
-                fiber_swap_log_state(f);
+                Fiber *fiber = fiber_get(f);
+                fiber_swap_log_state(fiber);
                 fiber_ops_set(NULL);
         }
         fiber_set_current(NULL);
 }
 
-static sd_event_source* fiber_current_event_source(Fiber *f) {
-        assert(f);
-        assert(f->state != FIBER_STATE_COMPLETED);
-        assert(f->event);
+static sd_event_source* fiber_current_event_source(sd_future *f) {
+        Fiber *fiber = fiber_get(f);
+        assert(fiber->state != FIBER_STATE_COMPLETED);
 
-        return sd_event_get_state(f->event) == SD_EVENT_EXITING ? f->exit_event_source : f->defer_event_source;
+        /* SD_EVENT_EXITING only covers the exit callback itself, not the time between callbacks or
+         * io_uring completions dispatched while exiting. The exit request persists across all of these. */
+        return sd_event_get_exit_code(sd_future_get_event(f), /* ret= */ NULL) >= 0 ?
+                fiber->exit_event_source : fiber->defer_event_source;
 }
 
 static int atfork_ret;
@@ -298,14 +306,12 @@ static void install_atfork(void) {
 }
 
 static void fiber_resolve(sd_future *f) {
-        Fiber *fiber = ASSERT_PTR(sd_future_get_private(ASSERT_PTR(f)));
+        Fiber *fiber = fiber_get(f);
 
         fiber->defer_event_source = sd_event_source_disable_unref(fiber->defer_event_source);
         fiber->exit_event_source = sd_event_source_disable_unref(fiber->exit_event_source);
-        /* The floating self-ref (if any) is potentially the last ref keeping the fiber alive — moving it
-         * into a local _cleanup_ slot ensures sd_future_resolve() runs callbacks and waiters while f is
-         * still valid; the local's cleanup drops the ref afterwards, at which point no further f->...
-         * access can happen. */
+        /* The floating reference may be the last one. Keep it until sd_future_resolve() has marked the
+         * future resolved and armed its callbacks; dropping it can free the fiber immediately. */
         _unused_ _cleanup_(sd_future_unrefp) sd_future *floating = TAKE_PTR(fiber->floating);
         sd_future_resolve(f, fiber->result);
 }
@@ -318,18 +324,23 @@ static const FiberOps fiber_ops = {
         .cancel_wait_unref = sd_future_cancel_wait_unref,
 };
 
-static void fiber_enter(Fiber *fiber, Fiber *prev, void **fake_stack_save) {
-        fiber_set_current(fiber);
+static void fiber_enter(sd_future *f, sd_future *prev, void **fake_stack_save) {
+        Fiber *fiber = fiber_get(f);
+        Fiber *prev_fiber = fiber_get_maybe(prev);
+
+        fiber_set_current(f);
         fiber_swap_log_state(fiber);
         if (!prev)
                 fiber_ops_set(&fiber_ops);
 
         struct iovec fiber_stack = fiber_stack_usable(&fiber->stack);
         start_switch_stack(fake_stack_save, &fiber_stack);
-        fiber_set_resume_stack(fiber, prev);
+        fiber_set_resume_stack(fiber, prev_fiber);
 }
 
-static void fiber_leave(Fiber *fiber, Fiber *prev, void *fake_stack_save) {
+static void fiber_leave(sd_future *f, sd_future *prev, void *fake_stack_save) {
+        Fiber *fiber = fiber_get(f);
+
         finish_switch_stack(fake_stack_save);
         if (!prev)
                 fiber_ops_set(NULL);
@@ -338,13 +349,13 @@ static void fiber_leave(Fiber *fiber, Fiber *prev, void *fake_stack_save) {
 }
 
 static int fiber_run(sd_future *f) {
-        Fiber *fiber = ASSERT_PTR(sd_future_get_private(ASSERT_PTR(f)));
+        Fiber *fiber = fiber_get(f);
         int r;
 
         if (fiber->state == FIBER_STATE_COMPLETED)
                 return -ESTALE;
 
-        assert(IN_SET(fiber->state, FIBER_STATE_INITIAL, FIBER_STATE_READY, FIBER_STATE_CANCELLED));
+        assert(IN_SET(fiber->state, FIBER_STATE_INITIAL, FIBER_STATE_READY));
 
         static pthread_once_t atfork_once = PTHREAD_ONCE_INIT;
         r = pthread_once(&atfork_once, install_atfork);
@@ -362,18 +373,23 @@ static int fiber_run(sd_future *f) {
          * completes. This matters when fiber_run() is invoked from within another fiber (e.g. an
          * sd-event dispatch that happens to be running inside a fiber context itself): the
          * LOG_SET_PREFIX/LOG_CONTEXT_PUSH above attached to whichever fiber was current at that moment,
-         * and their scope-level cleanup must see the same fiber_get_current() when it runs to detach
+         * and their scope-level cleanup must see the same sd_fiber_get_current() when it runs to detach
          * them from the correct list. */
-        Fiber *prev = fiber_get_current();
+        sd_future *prev = sd_fiber_get_current();
         void *fake_stack_save = NULL;
-        fiber_enter(fiber, prev, &fake_stack_save);
+
+        /* INITIAL means the function has never started; READY means it yielded or was woken. Neither
+         * includes execution: stay RUNNING until a real yield, suspension, or completion, even while
+         * a nested event loop dispatches another fiber and temporarily changes the current fiber. */
+        fiber->state = FIBER_STATE_RUNNING;
+        fiber_enter(f, prev, &fake_stack_save);
 
         /* This is where we start executing the fiber. Once it yields, we continue here as if nothing
          * happened. resume_context captures this point; the fiber siglongjmps back to it. */
         if (sigsetjmp(fiber->resume_context, 0) == 0)
                 siglongjmp_unchecked(fiber->context, 1);
 
-        fiber_leave(fiber, prev, fake_stack_save);
+        fiber_leave(f, prev, fake_stack_save);
 
         switch (fiber->state) {
 
@@ -386,11 +402,10 @@ static int fiber_run(sd_future *f) {
                 fiber_resolve(f);
                 break;
 
-        case FIBER_STATE_CANCELLED:
         case FIBER_STATE_READY:
                 log_debug("Fiber yielded execution");
 
-                r = sd_event_source_set_enabled(fiber_current_event_source(fiber), SD_EVENT_ONESHOT);
+                r = sd_event_source_set_enabled(fiber_current_event_source(f), SD_EVENT_ONESHOT);
                 if (r < 0)
                         return r;
                 break;
@@ -408,33 +423,34 @@ static int fiber_run(sd_future *f) {
 }
 
 static int fiber_cancel(sd_future *f) {
-        Fiber *fiber = ASSERT_PTR(sd_future_get_private(ASSERT_PTR(f)));
+        Fiber *fiber = fiber_get(f);
         int r;
 
-        assert(fiber != fiber_get_current());
+        assert(f != sd_fiber_get_current());
 
-        if (IN_SET(fiber->state, FIBER_STATE_COMPLETED, FIBER_STATE_CANCELLED))
+        if (fiber->state == FIBER_STATE_COMPLETED)
                 return 0;
 
         if (fiber->state == FIBER_STATE_INITIAL) {
-                /* The fiber's stack was allocated but never entered, so there are no scope-level cleanups
-                 * waiting to run. Skip the dispatch round-trip that would just have fiber_entry_point()
-                 * fall straight through with -ECANCELED, and settle the future right here — mirroring the
-                 * FIBER_STATE_COMPLETED branch of fiber_run(). */
+                /* The fiber's function has never started, so there are no scope-level cleanups
+                 * waiting to run. Skip the dispatch round-trip and settle the future right here —
+                 * mirroring the FIBER_STATE_COMPLETED branch of fiber_run(). */
                 fiber->result = -ECANCELED;
                 fiber->state = FIBER_STATE_COMPLETED;
                 fiber_resolve(f);
                 return 1;
         }
 
-        /* Once we cancel a fiber, we want to immediately resume it with -ECANCELED. */
-        r = sd_event_source_set_enabled(fiber_current_event_source(fiber), SD_EVENT_ONESHOT);
+        bool queued = fiber->result_pending && fiber->result == -ECANCELED;
+
+        /* Even an already-queued cancellation may need to move a pending defer dispatch to the exit
+         * source. Let sd_fiber_resume() handle scheduling in both cases, while keeping cancellation
+         * idempotent. -ECANCELED is sticky, so an async wakeup cannot overwrite it. */
+        r = sd_fiber_resume(f, -ECANCELED);
         if (r < 0)
                 return r;
 
-        fiber->state = FIBER_STATE_CANCELLED;
-
-        return 1;
+        return !queued;
 }
 
 static int fiber_on_defer(sd_event_source *s, void *userdata) {
@@ -444,21 +460,25 @@ static int fiber_on_defer(sd_event_source *s, void *userdata) {
 
 static int fiber_on_exit(sd_event_source *s, void *userdata) {
         sd_future *f = ASSERT_PTR(userdata);
-        Fiber *fiber = ASSERT_PTR(sd_future_get_private(f));
+        Fiber *fiber = fiber_get(f);
         int r;
 
-        /* The fiber may already have completed via the regular defer path before sd_event_exit()
-         * fires the exit source; in that case there's nothing left to drive and we'd otherwise
-         * trip fiber_run()'s -ESTALE return, which sd_event would log spuriously and disable the
-         * source for. */
-        if (fiber->state == FIBER_STATE_COMPLETED)
-                return 0;
+        /* An INITIAL fiber is cancelled synchronously; a COMPLETED fiber needs no further work.
+         * Return directly: cancelling an unstarted floating fiber may drop its last self-reference,
+         * so neither f nor fiber may be accessed afterwards. */
+        if (IN_SET(fiber->state, FIBER_STATE_INITIAL, FIBER_STATE_COMPLETED))
+                return fiber_cancel(f);
 
-        /* If fiber_cancel() returned 1 the fiber was just marked cancelled and its deferred/exit event
-         * source was re-armed; we let the event loop dispatch that source on the next iteration so it goes
-         * through the normal fiber_on_defer/fiber_on_exit path rather than running it recursively here. */
+        /* Cancellation of a started fiber only queues an interruption; it cannot resolve it here. */
         r = fiber_cancel(f);
-        if (r != 0)
+        if (r < 0)
+                return r;
+
+        /* Run directly in this dispatch. Cancellation re-arms the source for a READY or SUSPENDED
+         * fiber; clear that redundant dispatch so cleanup can suspend without being run again
+         * prematurely. fiber_run() will re-arm the source itself if the fiber yields. */
+        r = sd_event_source_set_enabled(s, SD_EVENT_OFF);
+        if (r < 0)
                 return r;
 
         return fiber_run(f);
@@ -469,7 +489,7 @@ static void* fiber_alloc(void) {
 }
 
 static void fiber_free(sd_future *f) {
-        Fiber *fiber = ASSERT_PTR(sd_future_get_private(f));
+        Fiber *fiber = fiber_get(f);
 
         /* To make sure all memory is deallocated, the fiber has to have completed by the time we free it to
          * make sure its stack has finished unwinding (which will invoke the registered cleanup functions).
@@ -480,7 +500,7 @@ static void fiber_free(sd_future *f) {
          * outer fiber should take care of cleaning up any created child fibers (for example using
          * sd_future_cancel_wait_unref()).
          *
-         * FIBER_STATE_INITIAL is also accepted: the stack was allocated but never entered, so there are no
+         * FIBER_STATE_INITIAL is also accepted: the function has never started, so there are no
          * registered cleanups to run. This covers the partial-construction failure path in sd_fiber_new()
          * as well as fibers that are unrefed before the event loop ever dispatches them. */
         assert(IN_SET(fiber->state, FIBER_STATE_INITIAL, FIBER_STATE_COMPLETED));
@@ -498,79 +518,81 @@ static void fiber_free(sd_future *f) {
 
         sd_event_source_disable_unref(fiber->defer_event_source);
         sd_event_source_disable_unref(fiber->exit_event_source);
-        sd_event_unref(fiber->event);
 
         free(fiber->name);
         free(fiber);
 }
 
 sd_future* sd_fiber_get_current(void) {
-        Fiber *f = fiber_get_current();
-        if (!f)
-                return NULL;
-
-        return sd_event_source_get_userdata(fiber_current_event_source(f));
+        return current_fiber;
 }
 
 int sd_fiber_is_running(void) {
-        return !!fiber_get_current();
+        return !!current_fiber;
 }
 
 sd_event* sd_fiber_get_event(void) {
-        Fiber *f = fiber_get_current();
+        sd_future *f = sd_fiber_get_current();
         assert_return(f, NULL);
-        return f->event;
+        return sd_future_get_event(f);
 }
 
 int sd_fiber_get_priority(int64_t *ret) {
-        Fiber *f = fiber_get_current();
+        sd_future *f = sd_fiber_get_current();
 
         assert_return(ret, -EINVAL);
         assert_return(f, -ESRCH);
 
-        *ret = f->priority;
+        Fiber *fiber = fiber_get(f);
+        *ret = fiber->priority;
         return 0;
 }
 
 static int fiber_swap(FiberState state) {
-        Fiber *f = ASSERT_PTR(fiber_get_current());
+        Fiber *fiber = fiber_get(sd_fiber_get_current());
 
-        f->state = state;
+        assert(fiber->state == FIBER_STATE_RUNNING);
+        assert(IN_SET(state, FIBER_STATE_READY, FIBER_STATE_SUSPENDED));
 
-        void *fake_stack_save = NULL;
+        /* A value queued by sd_fiber_resume() while the fiber was running short-circuits the swap:
+         * deliver it as if we had suspended and been resumed instantly, without round-tripping
+         * through the event loop. Neither yield nor suspend changes the scheduling state in this case:
+         * the caller must first observe the queued result before parking or yielding again. */
+        if (!fiber->result_pending) {
+                fiber->state = state;
 
-        if (sigsetjmp(f->context, 0) == 0) {
-                start_switch_stack(&fake_stack_save, &f->resume_stack);
-                siglongjmp_unchecked(f->resume_context, 1);
+                void *fake_stack_save = NULL;
+
+                if (sigsetjmp(fiber->context, 0) == 0) {
+                        start_switch_stack(&fake_stack_save, &fiber->resume_stack);
+                        siglongjmp_unchecked(fiber->resume_context, 1);
+                }
+
+                finish_switch_stack(fake_stack_save);
         }
 
-        finish_switch_stack(fake_stack_save);
+        assert(fiber->state == FIBER_STATE_RUNNING);
 
-        /* When we get here, we've been resumed. */
-
-        if (f->state == FIBER_STATE_CANCELLED)
-                return -ECANCELED;
-
-        /* sd_fiber_resume() stashes the resumer's value (an async wakeup error from a deadline
-         * timer, an io_uring CQE result, etc.) into f->result for us to surface here. Consume it
-         * unconditionally so it doesn't pollute subsequent suspends or the fiber's eventual return
-         * value — both negative errors and positive payloads (byte counts, accepted fds, revents
-         * masks) are valid resume values. */
-        return TAKE_GENERIC(f->result, int, 0);
+        /* Consume the resume value after either path, whether it was queued while running or
+         * delivered after a context switch. Clear it unconditionally so it doesn't pollute
+         * subsequent suspends or the fiber's eventual return value — both negative errors and
+         * positive payloads (byte counts, accepted fds, revents masks) are valid resume values. */
+        fiber->result_pending = false;
+        return TAKE_GENERIC(fiber->result, int, 0);
 }
 
 int sd_fiber_yield(void) {
-        assert_return(fiber_get_current(), -ESRCH);
+        assert_return(sd_fiber_get_current(), -ESRCH);
         return fiber_swap(FIBER_STATE_READY);
 }
 
 int sd_fiber_suspend(void) {
-        assert_return(fiber_get_current(), -ESRCH);
+        assert_return(sd_fiber_get_current(), -ESRCH);
         return fiber_swap(FIBER_STATE_SUSPENDED);
 }
 
 static int fiber_set_priority(sd_future *f, int64_t priority) {
-        Fiber *fiber = ASSERT_PTR(sd_future_get_private(ASSERT_PTR(f)));
+        Fiber *fiber = fiber_get(f);
         int r = 0;
 
         if (fiber->defer_event_source)
@@ -588,22 +610,49 @@ static int fiber_set_priority(sd_future *f, int64_t priority) {
 static const sd_future_ops fiber_future_ops;
 
 int sd_fiber_resume(sd_future *f, int result) {
+        int r;
+
         assert_return(f, -EINVAL);
         assert_return(sd_future_get_ops(f) == &fiber_future_ops, -EINVAL);
 
-        Fiber *fiber = ASSERT_PTR(sd_future_get_private(f));
+        Fiber *fiber = fiber_get(f);
 
-        if (fiber->state != FIBER_STATE_SUSPENDED)
+        if (fiber->state == FIBER_STATE_COMPLETED)
                 return 0;
 
-        /* Stash the result so fiber_swap() returns it from sd_fiber_suspend(). */
+        /* Cancellation outranks timeout; neither may be overwritten by an ordinary wakeup.
+         * Make sure we always schedule below as a retained result may need to be moved from the
+         * defer source to the exit source on event loop exit. */
+        if (fiber->result_pending &&
+            (fiber->result == -ECANCELED || (fiber->result == -ETIME && result != -ECANCELED)))
+                result = fiber->result;
+
         fiber->result = result;
+        fiber->result_pending = true;
+
+        /* INITIAL already has a dispatch queued. RUNNING fibers (including active ancestors in nested
+         * loops) consume the result at their next fiber_swap(), without needing another dispatch. */
+        if (IN_SET(fiber->state, FIBER_STATE_INITIAL, FIBER_STATE_RUNNING))
+                return 0;
+
+        assert(IN_SET(fiber->state, FIBER_STATE_READY, FIBER_STATE_SUSPENDED));
+
+        /* READY may need moving from defer source to exit source scheduling. Arm before changing state
+         * so a failure cannot leave a suspended fiber marked READY without a dispatch. */
+        sd_event_source *source = fiber_current_event_source(f);
+        r = sd_event_source_set_enabled(source, SD_EVENT_ONESHOT);
+        if (r < 0)
+                return r;
+
         fiber->state = FIBER_STATE_READY;
-        return sd_event_source_set_enabled(fiber_current_event_source(fiber), SD_EVENT_ONESHOT);
+        if (source == fiber->exit_event_source)
+                return sd_event_source_set_enabled(fiber->defer_event_source, SD_EVENT_OFF);
+
+        return 0;
 }
 
-/* The fiber_future ops pass the Fiber pointer through as the future's private state. The fiber resolves
- * its own future once it finishes running, so fiber_cancel() intentionally does not resolve. */
+/* The fiber_future ops pass the Fiber pointer through as the future's private state. Once started, the
+ * fiber resolves its own future when it finishes running; cancellation only queues an interruption. */
 static const sd_future_ops fiber_future_ops = {
         .size = sizeof(sd_future_ops),
         .alloc = fiber_alloc,
@@ -622,12 +671,12 @@ int sd_fiber_new(sd_event *e, const char *name, sd_fiber_func_t func, void *user
         if (IN_SET(sd_event_get_state(e), SD_EVENT_EXITING, SD_EVENT_FINISHED))
                 return -ECANCELED;
 
-        _cleanup_(sd_future_unrefp) sd_future *f = NULL;
-        r = sd_future_new(&fiber_future_ops, &f);
+        _cleanup_(sd_future_cancel_unrefp) sd_future *f = NULL;
+        r = sd_future_new(e, &fiber_future_ops, &f);
         if (r < 0)
                 return r;
 
-        Fiber *fiber = ASSERT_PTR(sd_future_get_private(f));
+        Fiber *fiber = fiber_get(f);
 
         struct rlimit rl = { RLIM_INFINITY, RLIM_INFINITY };
         if (getrlimit(RLIMIT_STACK, &rl) < 0)
@@ -647,7 +696,6 @@ int sd_fiber_new(sd_event *e, const char *name, sd_fiber_func_t func, void *user
                 .name = strdup(name),
                 .func = func,
                 .userdata = userdata,
-                .event = sd_event_ref(e),
         };
         if (!fiber->name)
                 return -ENOMEM;
@@ -665,7 +713,7 @@ int sd_fiber_new(sd_event *e, const char *name, sd_fiber_func_t func, void *user
                         (uint8_t*) usable.iov_base + usable.iov_len);
 #endif
 
-        r = fiber_init(fiber);
+        r = fiber_init(f);
         if (r < 0)
                 return r;
 
@@ -708,6 +756,9 @@ int sd_fiber_new(sd_event *e, const char *name, sd_fiber_func_t func, void *user
                 r = sd_fiber_set_floating(f, true);
                 if (r < 0)
                         return r;
+
+                /* Floating self-ref keeps the fiber alive; release our local ref without cancelling. */
+                f = sd_future_unref(f);
         }
 
         /* We only take ownership of the given userdata pointer on success so assign the destroy callback
@@ -721,7 +772,7 @@ int sd_fiber_set_floating(sd_future *f, int b) {
         assert_return(f, -EINVAL);
         assert_return(sd_future_get_ops(f) == &fiber_future_ops, -EINVAL);
 
-        Fiber *fiber = ASSERT_PTR(sd_future_get_private(f));
+        Fiber *fiber = fiber_get(f);
 
         if (!!fiber->floating == !!b)
                 return 0;
@@ -741,12 +792,12 @@ int sd_fiber_get_floating(sd_future *f) {
         assert_return(f, -EINVAL);
         assert_return(sd_future_get_ops(f) == &fiber_future_ops, -EINVAL);
 
-        Fiber *fiber = ASSERT_PTR(sd_future_get_private(f));
+        Fiber *fiber = fiber_get(f);
         return !!fiber->floating;
 }
 
 int sd_fiber_sleep(uint64_t usec) {
-        Fiber *f = fiber_get_current();
+        sd_future *f = sd_fiber_get_current();
         int r;
 
         if (!f)
@@ -760,11 +811,9 @@ int sd_fiber_sleep(uint64_t usec) {
         if (usec == USEC_INFINITY)
                 return sd_fiber_suspend();
 
-        assert(f->event);
-
         _cleanup_(sd_future_cancel_wait_unrefp) sd_future *timer = NULL;
         r = future_new_time_relative(
-                        f->event,
+                        sd_future_get_event(f),
                         CLOCK_MONOTONIC,
                         usec,
                         /* accuracy= */ 1,
@@ -773,7 +822,33 @@ int sd_fiber_sleep(uint64_t usec) {
         if (r < 0)
                 return r;
 
-        return sd_fiber_suspend();
+        return sd_fiber_await(timer);
+}
+
+static int fiber_wait_callback(sd_future *target, void *userdata) {
+        /* Completion only wakes the waiter. Passing the future's error here would make it
+         * indistinguishable from an interruption of the waiting fiber. */
+        return sd_fiber_resume(userdata, 0);
+}
+
+static int fiber_suspend_for(sd_future *f, sd_future *target) {
+        Fiber *fiber = fiber_get(f);
+        int r;
+
+        assert(!fiber->awaiting);
+
+        fiber->awaiting = target;
+        r = sd_fiber_suspend();
+        fiber->awaiting = NULL;
+
+        return r;
+}
+
+sd_future* sd_fiber_get_awaiting(sd_future *f) {
+        assert_return(f, NULL);
+        assert_return(sd_future_get_ops(f) == &fiber_future_ops, NULL);
+
+        return fiber_get(f)->awaiting;
 }
 
 int sd_fiber_await(sd_future *target) {
@@ -783,37 +858,120 @@ int sd_fiber_await(sd_future *target) {
         assert_return(f, -ESRCH);
         assert_return(target, -EINVAL);
         assert_return(target != f, -EDEADLK);
-
-        Fiber *fiber = ASSERT_PTR(sd_future_get_private(f));
+        assert_return(sd_future_get_event(target) == sd_future_get_event(f), -EINVAL);
 
         if (sd_future_state(target) == SD_FUTURE_RESOLVED)
                 return sd_future_result(target);
 
-        /* Note that we do allow waiting for other fibers when the event loop is exiting, since waiting for
-         * other fibers does not require adding new event sources to the event loop. */
-        if (sd_event_get_state(fiber->event) == SD_EVENT_FINISHED)
-                return -ECANCELED;
+        /* The fiber is executing inside event dispatch, so its loop cannot have finished yet.
+         * Waiting during exit is supported by the slot's exit source. */
+        assert(sd_event_get_state(sd_future_get_event(f)) != SD_EVENT_FINISHED);
 
-        _cleanup_(sd_future_cancel_wait_unrefp) sd_future *wait = NULL;
-        r = sd_future_new_wait(target, &wait);
+        _cleanup_(sd_future_slot_unrefp) sd_future_slot *slot = NULL;
+        r = sd_future_add_callback(target, &slot, fiber_wait_callback, f);
         if (r < 0)
                 return r;
 
-        return sd_fiber_suspend();
+        r = fiber_suspend_for(f, target);
+        if (r < 0)
+                return r;
+
+        if (sd_future_state(target) != SD_FUTURE_RESOLVED)
+                return -EBUSY;
+
+        /* Read the result while the slot still pins the target: releasing it may drop the last ref. */
+        return sd_future_result(target);
+}
+
+sd_future* sd_future_cancel_wait_unref(sd_future *f) {
+        sd_future *self = sd_fiber_get_current();
+        int r, q = 0;
+
+        if (!f)
+                return NULL;
+
+        /* Outside a fiber we can still clean up futures whose cancellation is synchronous, or which
+         * are already resolved. Asynchronous cancellation requires a fiber to await completion. */
+        if (!self)
+                return sd_future_cancel_unref(f);
+
+        /* Invalid calls still consume their reference. Releasing the last reference to a pending
+         * future remains a programming error, diagnosed by sd_future_free(). */
+        assert_return(f != self, sd_future_unref(f));
+
+        for (;;) {
+                r = sd_future_cancel(f);
+                if (r < 0)
+                        log_debug_errno(r, "Failed to cancel future, ignoring: %m");
+
+                if (sd_future_state(f) != SD_FUTURE_PENDING)
+                        break;
+
+                /* Synchronous cancellation needs no dispatch and works across event loops too. */
+                assert_return(sd_future_get_event(f) == sd_future_get_event(self), sd_future_unref(f));
+
+                /* Unlike sd_fiber_await(), a failure to set up the wait must be told apart from a
+                 * wait that was interrupted or woken spuriously. Retrying the former would spin
+                 * without ever yielding to the event loop, so nothing could resolve the future. */
+                _cleanup_(sd_future_slot_unrefp) sd_future_slot *slot = NULL;
+                r = sd_future_add_callback(f, &slot, fiber_wait_callback, self);
+                if (r < 0) {
+                        log_debug_errno(r, "Failed to wait for future to finish, giving up: %m");
+                        break;
+                }
+
+                r = fiber_suspend_for(self, f);
+
+                if (sd_future_state(f) == SD_FUTURE_RESOLVED) {
+                        int fr = sd_future_result(f);
+                        if (fr < 0 && fr != -ECANCELED)
+                                log_debug_errno(fr, "Future resolved with error, ignoring: %m");
+                }
+
+                if (IN_SET(r, -ECANCELED, -ETIME)) {
+                        log_debug_errno(r, "Interrupted while waiting for future to finish, deferring the interruption until the future resolves: %m");
+                        /* The wait reports interruptions separately from the future's result. Preserve
+                         * them even if the future resolved concurrently, or a previous cleanup queued
+                         * the interruption before this wait. Cancellation takes precedence over timeout. */
+                        if (q != -ECANCELED)
+                                q = r;
+                } else if (r < 0)
+                        log_debug_errno(r, "Failed to wait for future to finish, ignoring: %m");
+
+                if (sd_future_state(f) != SD_FUTURE_PENDING)
+                        break;
+        }
+
+        /* Re-queue the interruption we held back, if any. */
+        if (q < 0) {
+                r = sd_fiber_resume(self, q);
+                if (r < 0)
+                        log_debug_errno(r, "Failed to re-queue interruption (%i) on calling fiber, ignoring: %m", q);
+        }
+
+        return sd_future_unref(f);
+}
+
+DEFINE_POINTER_ARRAY_CLEAR_FUNC(sd_future*, sd_future_cancel_wait_unref);
+DEFINE_POINTER_ARRAY_FREE_FUNC(sd_future*, sd_future_cancel_wait_unref);
+
+static int fiber_timeout_callback(sd_future *f, void *userdata) {
+        /* Timeout scopes deliver -ETIME, unlike ordinary waits which only signal completion. */
+        return sd_fiber_resume(userdata, sd_future_result(f));
 }
 
 sd_future* sd_fiber_timeout(uint64_t timeout) {
-        Fiber *fiber = fiber_get_current();
+        sd_future *f = sd_fiber_get_current();
         int r;
 
-        assert_return(fiber, NULL);
+        assert_return(f, NULL);
 
         if (timeout == USEC_INFINITY)
                 return NULL;
 
-        sd_future *timer;
+        _cleanup_(sd_future_cancel_wait_unrefp) sd_future *timer = NULL;
         r = future_new_time_relative(
-                        fiber->event,
+                        sd_future_get_event(f),
                         CLOCK_MONOTONIC,
                         timeout,
                         /* accuracy= */ 1,
@@ -823,5 +981,12 @@ sd_future* sd_fiber_timeout(uint64_t timeout) {
                 return NULL; /* On allocation failure no timer is armed and the scope becomes a no-op.
                               * Errors here are rare; if the caller cares they can compare to NULL. */
 
-        return timer;
+        /* The whole point of SD_FIBER_TIMEOUT is to wake the calling fiber when the deadline
+         * fires (so a later sd_fiber_suspend / sd_fiber_await returns -ETIME from this timer's
+         * resolve). Install a floating resume callback bound to the timer's lifetime. */
+        r = sd_future_add_callback(timer, /* ret_slot= */ NULL, fiber_timeout_callback, f);
+        if (r < 0)
+                return NULL;
+
+        return TAKE_PTR(timer);
 }

src/libsystemd/sd-future/future-group.c

@@ -0,0 +1,322 @@
+/* SPDX-License-Identifier: LGPL-2.1-or-later */
+
+#include <stdarg.h>
+
+#include "sd-future.h"
+
+#include "alloc-util.h"
+#include "errno-util.h"
+#include "macro.h"
+
+typedef struct FutureGroup {
+        uint64_t policy;
+
+        /* An explicit priority applies to existing and future children. Otherwise preserve theirs. */
+        int64_t priority;
+        bool priority_set;
+
+        sd_future_slot **slots;
+        size_t n_slots;
+
+        /* The fiber the group was created on, if it uses the same event loop. When the
+         * group settles on an error and IGNORE_ERRORS is unset, this fiber is cancelled so it
+         * notices the failure even if it hasn't started awaiting the group (a child error cancels
+         * the parent). parent_slot's callback NULLs `parent` if the parent resolves before the
+         * group does. */
+        sd_future *parent;
+        sd_future_slot *parent_slot;
+
+        /* Set once future_group_finalize() has been entered. The outcome is decided (stored in
+         * `result`) and the group is "draining" — waiting for any still-pending children to
+         * actually settle before we resolve. While set, the result cannot change and add
+         * rejects with -ESTALE. */
+        bool finalizing;
+        int result;
+} FutureGroup;
+
+static void* future_group_alloc(void) {
+        return new0(FutureGroup, 1);
+}
+
+static void future_group_free(sd_future *f) {
+        FutureGroup *fg = ASSERT_PTR(sd_future_get_private(f));
+
+        sd_future_slot_unref(fg->parent_slot);
+        FOREACH_ARRAY(slot_p, fg->slots, fg->n_slots)
+                sd_future_slot_unref(*slot_p);
+        free(fg->slots);
+        free(fg);
+}
+
+static int future_group_parent_resolved(sd_future *parent, void *userdata) {
+        FutureGroup *fg = ASSERT_PTR(userdata);
+        fg->parent = NULL;
+        fg->parent_slot = sd_future_slot_unref(fg->parent_slot);
+        return 0;
+}
+
+static int future_group_check(sd_future *g);
+
+static int future_group_finalize(sd_future *g, int result, bool propagate_error) {
+        FutureGroup *fg = ASSERT_PTR(sd_future_get_private(g));
+        int r = 0;
+
+        if (fg->finalizing)
+                /* Outcome already locked: ignore subsequent attempts. Mirrors the old "group
+                 * is already RESOLVED, so further cancels are no-ops" behaviour. */
+                return 0;
+
+        fg->finalizing = true;
+        fg->result = result;
+
+        /* A cancellation error does not make a pending child safe to release. Keep waiting for its
+         * resolution even when cancellation is unsupported or fails; child implementations must still
+         * arrange completion before the group can finish draining. */
+        FOREACH_ARRAY(slot_p, fg->slots, fg->n_slots) {
+                sd_future *child = sd_future_slot_get_future(*slot_p);
+                if (sd_future_state(child) == SD_FUTURE_PENDING)
+                        RET_GATHER(r, sd_future_cancel(child));
+        }
+
+        /* If we're settling because of a child error (and the user hasn't opted into ignoring
+         * errors), cancel the parent fiber so it notices the failure even if it hasn't
+         * started awaiting the group yet. An active await means the parent will receive the
+         * group's actual error, so cancelling it would only hide that error behind -ECANCELED.
+         * The suppression ends automatically when an interrupted wait returns; a peer's wait
+         * cannot suppress cancellation of the parent.
+         * Explicit group cancellation never propagates upward. */
+        if (propagate_error && result < 0 &&
+            !FLAGS_SET(fg->policy, SD_FUTURE_GROUP_IGNORE_ERRORS) &&
+            fg->parent &&
+            fg->parent != sd_fiber_get_current() &&
+            sd_fiber_get_awaiting(fg->parent) != g)
+                RET_GATHER(r, sd_future_cancel(fg->parent));
+
+        /* Re-check: if every child settled synchronously during the cancel loop the group can
+         * resolve now; otherwise wait for the group_child_resolved callbacks to drive the
+         * drain branch of check(). */
+        RET_GATHER(r, future_group_check(g));
+        return r;
+}
+
+static int future_group_check(sd_future *g) {
+        FutureGroup *fg = ASSERT_PTR(sd_future_get_private(g));
+
+        if (sd_future_state(g) == SD_FUTURE_RESOLVED)
+                return 0;
+
+        if (fg->finalizing) {
+                /* Outcome decided; resolve once every child has actually settled so callers
+                 * observing the group's resolution see every child in RESOLVED state. An empty
+                 * finalizing group resolves immediately (the FOREACH_ARRAY body never runs). */
+                FOREACH_ARRAY(slot_p, fg->slots, fg->n_slots)
+                        if (sd_future_state(sd_future_slot_get_future(*slot_p)) != SD_FUTURE_RESOLVED)
+                                return 0;
+                return sd_future_resolve(g, fg->result);
+        }
+
+        if (fg->n_slots == 0)
+                /* Empty group has nothing to wait for: leave it pending so the user can still
+                 * add children (or cancel the group). Otherwise an early set_policy on a
+                 * fresh group would settle it before any child got added. */
+                return 0;
+
+        bool wait_any = FLAGS_SET(fg->policy, SD_FUTURE_GROUP_WAIT_ANY);
+        bool ignore_errors = FLAGS_SET(fg->policy, SD_FUTURE_GROUP_IGNORE_ERRORS);
+
+        size_t n_resolved = 0;
+        int first_error = 0, first_success = 0;
+        bool any_success = false;
+
+        FOREACH_ARRAY(slot_p, fg->slots, fg->n_slots) {
+                sd_future *child = sd_future_slot_get_future(*slot_p);
+                if (sd_future_state(child) != SD_FUTURE_RESOLVED)
+                        continue;
+
+                n_resolved++;
+                int cr = sd_future_result(child);
+                if (cr < 0) {
+                        if (first_error == 0)
+                                first_error = cr;
+                } else if (!any_success) {
+                        any_success = true;
+                        first_success = cr;
+                }
+        }
+
+        bool all_done = (n_resolved == fg->n_slots);
+
+        int result;
+        if (wait_any && any_success)
+                result = first_success;        /* wait_any short-circuits on first success */
+        else if (!ignore_errors && first_error != 0)
+                result = first_error;          /* fail-fast on error unless ignored */
+        else if (all_done)
+                result = first_error;          /* everyone settled: 0 if no errors */
+        else
+                return 0;
+
+        return future_group_finalize(g, result, /* propagate_error= */ true);
+}
+
+static int future_group_cancel(sd_future *f) {
+        /* Explicit group cancellation affects its children, not the fiber that created the group. */
+        return future_group_finalize(f, -ECANCELED, /* propagate_error= */ false);
+}
+
+static int future_group_set_child_priority(sd_future *child, int64_t priority) {
+        int r;
+
+        r = sd_future_set_priority(child, priority);
+        /* Some children do not support priorities or have already resolved. */
+        if (r < 0 && !IN_SET(r, -EOPNOTSUPP, -ESTALE))
+                return r;
+
+        return 0;
+}
+
+static int future_group_set_priority(sd_future *f, int64_t priority) {
+        FutureGroup *fg = ASSERT_PTR(sd_future_get_private(f));
+        int r = 0;
+
+        FOREACH_ARRAY(slot_p, fg->slots, fg->n_slots)
+                RET_GATHER(r, future_group_set_child_priority(sd_future_slot_get_future(*slot_p), priority));
+        if (r < 0)
+                return r;
+
+        fg->priority = priority;
+        fg->priority_set = true;
+        return 0;
+}
+
+static const sd_future_ops future_group_ops = {
+        .size = sizeof(sd_future_ops),
+        .alloc = future_group_alloc,
+        .free = future_group_free,
+        .cancel = future_group_cancel,
+        .set_priority = future_group_set_priority,
+};
+
+int sd_future_group_new(sd_event *e, sd_future **ret) {
+        int r;
+
+        assert_return(e, -EINVAL);
+        assert_return(ret, -EINVAL);
+
+        _cleanup_(sd_future_cancel_unrefp) sd_future *g = NULL;
+        r = sd_future_new(e, &future_group_ops, &g);
+        if (r < 0)
+                return r;
+
+        sd_future *parent = sd_fiber_get_current();
+        if (parent && sd_future_get_event(parent) == sd_future_get_event(g)) {
+                FutureGroup *fg = sd_future_get_private(g);
+                r = sd_future_add_callback(parent, &fg->parent_slot, future_group_parent_resolved, fg);
+                if (r < 0)
+                        return r;
+                fg->parent = parent;
+        }
+
+        *ret = TAKE_PTR(g);
+        return 0;
+}
+
+size_t sd_future_group_size(sd_future *f) {
+        if (!f)
+                return 0;
+
+        assert_return(sd_future_get_ops(f) == &future_group_ops, 0);
+
+        FutureGroup *fg = sd_future_get_private(f);
+        return fg->n_slots;
+}
+
+int sd_future_group_set_policy(sd_future *f, uint64_t policy) {
+        assert_return(f, -EINVAL);
+        assert_return(sd_future_get_ops(f) == &future_group_ops, -EINVAL);
+        assert_return(sd_future_state(f) == SD_FUTURE_PENDING, -ESTALE);
+        assert_return((policy & ~(uint64_t) _SD_FUTURE_GROUP_POLICY_MASK) == 0, -EINVAL);
+
+        /* Policy must be configured before any children are added — once a child is in flight,
+         * the resolution mechanics are locked in. This keeps the API friction-free: callers
+         * don't have to reason about mid-flight reshuffling of which children get cancelled. */
+        FutureGroup *fg = sd_future_get_private(f);
+        assert_return(fg->n_slots == 0, -ESTALE);
+
+        fg->policy = policy;
+        return 0;
+}
+
+static int group_child_resolved(sd_future *child, void *userdata) {
+        sd_future *g = ASSERT_PTR(userdata);
+        return future_group_check(g);
+}
+
+int sd_future_group_add(sd_future *f, sd_future *child) {
+        int r;
+
+        assert_return(f, -EINVAL);
+        assert_return(child, -EINVAL);
+        assert_return(sd_future_get_ops(f) == &future_group_ops, -EINVAL);
+        assert_return(sd_future_state(f) == SD_FUTURE_PENDING, -ESTALE);
+        /* A group waiting for itself can never settle. */
+        assert_return(child != f, -EINVAL);
+
+        /* Child notifications are dispatched on the child's event loop. */
+        assert_return(sd_future_get_event(child) == sd_future_get_event(f), -EINVAL);
+
+        FutureGroup *fg = sd_future_get_private(f);
+        /* Group is draining: a freshly-added pending child would have missed the cancel loop
+         * and hang us forever waiting for it to settle. */
+        assert_return(!fg->finalizing, -ESTALE);
+
+        if (!GREEDY_REALLOC(fg->slots, fg->n_slots + 1))
+                return -ENOMEM;
+
+        _cleanup_(sd_future_slot_unrefp) sd_future_slot *slot = NULL;
+        r = sd_future_add_callback(child, &slot, group_child_resolved, f);
+        if (r < 0)
+                return r;
+
+        if (fg->priority_set) {
+                r = future_group_set_child_priority(child, fg->priority);
+                if (r < 0)
+                        return r;
+        }
+
+        fg->slots[fg->n_slots++] = TAKE_PTR(slot);
+
+        return 0;
+}
+
+int sd_future_group_add_many_internal(sd_future *f, ...) {
+        assert_return(f, -EINVAL);
+        assert_return(sd_future_get_ops(f) == &future_group_ops, -EINVAL);
+
+        FutureGroup *fg = sd_future_get_private(f);
+        size_t before = fg->n_slots;
+        int r = 0;
+
+        va_list ap;
+        va_start(ap, f);
+        for (;;) {
+                sd_future *child = va_arg(ap, sd_future*);
+                if (!child)
+                        break;
+
+                r = sd_future_group_add(f, child);
+                if (r < 0)
+                        break;
+        }
+        va_end(ap);
+
+        if (r < 0)
+                /* No callbacks run inline while adding children, so the group cannot start finalizing
+                 * during this call. Roll back only this call's additions. */
+                while (fg->n_slots > before) {
+                        sd_future_slot_unref(fg->slots[--fg->n_slots]);
+                        fg->slots[fg->n_slots] = NULL;
+                }
+
+        return r;
+}

src/libsystemd/sd-future/sd-future.c

@@ -1,130 +1,180 @@
 /* SPDX-License-Identifier: LGPL-2.1-or-later */
 
+#include "sd-event.h"
 #include "sd-future.h"
 
 #include "alloc-util.h"
 #include "errno-util.h"
+#include "event-future.h"
+#include "event-util.h"
 #include "log.h"
 #include "macro.h"
 #include "set.h"
 
+struct sd_future_slot {
+        unsigned n_ref;
+
+        /* Back-pointer to the future the slot is attached to.
+         *
+         * Ref ownership is asymmetric (same trick as sd_bus_slot/bus->slots): when the slot
+         * is non-floating the SLOT owns a ref on the future; when floating, the FUTURE owns
+         * a ref on the slot. So `slots` is always a borrowed pointer collection regardless.
+         *
+         * Consequence of the non-floating case: because the slot holds a ref, dropping the slot may
+         * perform the future's last unref — which, like any last unref, requires the future to already
+         * be RESOLVED. A caller that releases its future handle and relies on the slot to keep the
+         * future alive must drive it to resolution before dropping the slot. */
+        sd_future *future;
+        bool floating;
+
+        sd_future_func_t callback;
+        void *userdata;
+
+        sd_event_source *defer_source;
+        sd_event_source *exit_source;
+};
+
 struct sd_future {
         unsigned n_ref;
 
         int state;
         int result;
 
-        Set *waiters;
+        sd_event *event;
 
-        sd_future_func_t callback;
-        void *userdata;
+        Set *slots;
 
         const sd_future_ops *ops;
 
         /* Opaque per-future state owned by the future implementation (the code that called
-         * sd_future_new()). The ops vtable above receives this pointer in its callbacks, and
-         * external code can fetch it via sd_future_get_private(). */
+         * sd_future_new()). The ops callbacks and external code access this state via
+         * sd_future_get_private(). */
         void *private;
 };
 
-static int fiber_resume_trampoline(sd_future *f) {
-        /* The future's result is what the fiber should resume with. Impls choose the value at
-         * resolution time — e.g. a deadline timer resolves with -ETIME, a wait future resolves
-         * with the target's result, a normal IO/sleep future resolves with 0 on success. */
-        return sd_fiber_resume(sd_future_get_userdata(f), sd_future_result(f));
+static int slot_dispatch_handler(sd_event_source *src, void *userdata) {
+        sd_future_slot *s = ASSERT_PTR(userdata);
+        sd_future *f = s->future;
+        int r;
+
+        /* Invoked when the slot's chosen event source fires. Hold a self-ref on `f`
+         * across the callback: a floating slot lives in f->slots, so if the callback
+         * drops the last user-side ref to f, sd_future_free would iterate floating
+         * slots and unref us — pulling the rug out from under the `sd_future_slot_unref`
+         * below. Holding the ref keeps f alive until we've cleaned ourselves up; the
+         * cleanup at scope exit drops it, freeing f if no one else held a ref. */
+        bool floating = s->floating; /* capture: non-floating s may be freed by callback */
+        _unused_ _cleanup_(sd_future_unrefp) sd_future *self = sd_future_ref(f);
+
+        /* Both sources are armed so a queued callback survives an exit request. Disable both before
+         * invoking it, since the callback may free the slot or itself request exit. Disabling defer
+         * and exit sources is non-failing during valid dispatch. */
+        assert_se(sd_event_source_set_enabled(s->defer_source, SD_EVENT_OFF) >= 0);
+        assert_se(sd_event_source_set_enabled(s->exit_source, SD_EVENT_OFF) >= 0);
+
+        r = s->callback(f, s->userdata);
+        if (floating)
+                sd_future_slot_unref(s);
+        return r;
+}
+
+static int slot_arm(sd_future_slot *s) {
+        int r = 0;
+
+        /* Exit may be requested after arming but before dispatching the callback. */
+        RET_GATHER(r, sd_event_source_set_enabled(s->defer_source, SD_EVENT_ONESHOT));
+        RET_GATHER(r, sd_event_source_set_enabled(s->exit_source, SD_EVENT_ONESHOT));
+        if (r < 0)
+                return log_debug_errno(r, "Failed to arm future callback: %m");
+
+        return 0;
 }
 
 int sd_future_resolve(sd_future *f, int result) {
         int r = 0;
 
         assert_return(f, -EINVAL);
-
-        if (f->state != SD_FUTURE_PENDING)
-                return 0;
-
-        /* Hold a self-ref across callback/waiter dispatch: callbacks (e.g. bus_fiber_resolved()
-         * dropping the tracking-set's ref) may legitimately release what would otherwise be the
-         * last reference, and we still access f->waiters below. The cleanup unrefs at scope exit,
-         * which is when freeing is safe again. */
-        _unused_ _cleanup_(sd_future_unrefp) sd_future *self = sd_future_ref(f);
+        assert_return(f->state == SD_FUTURE_PENDING, -ESTALE);
 
         f->state = SD_FUTURE_RESOLVED;
         f->result = result;
 
-        if (f->callback)
-                RET_GATHER(r, f->callback(f));
-
-        /* We'd like the set to not be modified while iterating over it, hence take ownership over it in
-         * a local variable. Otherwise code invoked via sd_future_resolve() could try to modify the set while
-         * we're iterating over it (for example wait_future_free()). */
-        Set *waiters = TAKE_PTR(f->waiters);
-        sd_future *w;
-        SET_FOREACH(w, waiters)
-                RET_GATHER(r, sd_future_resolve(w, result));
-
-        set_free(waiters);
+        /* Enable each slot's event sources so the callback fires on the next loop iteration. Deferring
+         * callbacks frees the resolver from reentrancy concerns, slot-set mutation during iteration,
+         * and callbacks dropping the future's last ref. */
+        sd_future_slot *s;
+        SET_FOREACH(s, f->slots)
+                RET_GATHER(r, slot_arm(s));
 
         return r;
 }
 
 static sd_future* sd_future_free(sd_future *f) {
-        if (!f)
-                return NULL;
+        /* By the time we tear down, the future must have reached a terminal state. Callers
+         * abandoning a still-PENDING future must drive it to RESOLVED first — typically via
+         * sd_future_cancel_unref() (non-fiber, synchronous-cancel impls) or
+         * sd_future_cancel_wait_unref() (fiber, awaits actual resolution). */
+        assert(f->state == SD_FUTURE_RESOLVED);
 
-        if (f->state == SD_FUTURE_PENDING)
-                sd_future_resolve(f, -ECANCELED);
-
-        set_free(f->waiters);
+        /* Any slot still in f->slots at this point must be floating: non-floating slots own
+         * a ref on f, so if any existed we wouldn't have reached free. (Slots can be added
+         * post-resolution via sd_future_add_callback — those are floating and may not have
+         * had their defer tick yet.) Tear them down by dropping the future's ref. */
+        Set *slots = TAKE_PTR(f->slots);
+        sd_future_slot *s;
+        SET_FOREACH(s, slots) {
+                assert(s->floating);
+                sd_future_slot_unref(s);
+        }
+        set_free(slots);
 
         if (f->ops->free)
                 f->ops->free(f);
 
+        sd_event_unref(f->event);
         return mfree(f);
 }
 
+/* Unref is a pure refcount op: dropping a ref does not resolve or cancel. If a caller wants
+ * pending work to complete (and floating callbacks to observe the outcome) before release,
+ * they must drive resolution explicitly — typically via sd_future_cancel() or
+ * sd_future_cancel_wait_unref(). Reaching sd_future_free() with state PENDING is a programming
+ * error and trips an assert: callers must drive the future to RESOLVED before dropping the last
+ * ref. */
 DEFINE_TRIVIAL_REF_UNREF_FUNC(sd_future, sd_future, sd_future_free);
+
 DEFINE_POINTER_ARRAY_CLEAR_FUNC(sd_future*, sd_future_unref);
 DEFINE_POINTER_ARRAY_FREE_FUNC(sd_future*, sd_future_unref);
 
-sd_future* sd_future_cancel_wait_unref(sd_future *f) {
+sd_future* sd_future_cancel_unref(sd_future *f) {
         int r;
 
         if (!f)
                 return NULL;
 
-        /* We have to be able to suspend until the fiber we're waiting for finishes, and that's only
-         * possible if we're running on a fiber ourselves. */
-        if (!sd_fiber_is_running())
-                return sd_future_unref(f);
-
+        /* Synchronous-cancel teardown for non-fiber contexts: drive the future to RESOLVED via
+         * ops->cancel, then drop the ref. Safe for impls whose ops->cancel resolves synchronously.
+         * For impls whose cancel is asynchronous, the future stays PENDING after this call and
+         * sd_future_free's strict assert will trip. Callers in that situation must use
+         * sd_future_cancel_wait_unref() from a fiber to await the actual resolution. */
         r = sd_future_cancel(f);
         if (r < 0)
                 log_debug_errno(r, "Failed to cancel future, ignoring: %m");
 
-        if (f->state == SD_FUTURE_PENDING) {
-                /* Fast path: when f's resolve callback already targets the current fiber (the default for
-                 * futures created on this fiber), we can suspend directly and let the existing trampoline
-                 * wake us up — no need to allocate a wait future just to learn about the resolution.
-                 * Otherwise fall back to sd_fiber_await() which sets up an explicit waiter. */
-                if (f->callback == fiber_resume_trampoline && f->userdata == sd_fiber_get_current())
-                        r = sd_fiber_suspend();
-                else
-                        r = sd_fiber_await(f);
-                if (r < 0 && r != -ECANCELED)
-                        log_debug_errno(r, "Failed to wait for future to finish, ignoring: %m");
-        }
-
         return sd_future_unref(f);
 }
 
-DEFINE_POINTER_ARRAY_CLEAR_FUNC(sd_future*, sd_future_cancel_wait_unref);
-DEFINE_POINTER_ARRAY_FREE_FUNC(sd_future*, sd_future_cancel_wait_unref);
+DEFINE_POINTER_ARRAY_CLEAR_FUNC(sd_future*, sd_future_cancel_unref);
+DEFINE_POINTER_ARRAY_FREE_FUNC(sd_future*, sd_future_cancel_unref);
 
-int sd_future_new(const sd_future_ops *ops, sd_future **ret) {
+int sd_future_new(sd_event *e, const sd_future_ops *ops, sd_future **ret) {
+        assert_return(e, -EINVAL);
+        assert_return(e = event_resolve(e), -ENOPKG);
         assert_return(ops, -EINVAL);
         assert_return(ops->size >= endoffsetof_field(sd_future_ops, set_priority), -EINVAL);
         assert_return(ops->alloc, -EINVAL);
         assert_return(ops->free, -EINVAL);
+        assert_return(ops->cancel, -EINVAL);
         assert_return(ret, -EINVAL);
 
         sd_future *f = new(sd_future, 1);
@@ -143,18 +193,16 @@ int sd_future_new(const sd_future_ops *ops, sd_future **ret) {
                 return -ENOMEM;
         }
 
-        /* If we're being created on a fiber, default the callback to resuming that fiber on resolve —
-         * this is almost always what you want, and it saves the usual set_callback boilerplate before
-         * sd_fiber_suspend(). Callers that want different behavior can override with
-         * sd_future_set_callback(). */
-        sd_future *fiber = sd_fiber_get_current();
-        if (fiber)
-                (void) sd_future_set_callback(f, fiber_resume_trampoline, fiber);
-
+        f->event = sd_event_ref(e);
         *ret = f;
         return 0;
 }
 
+sd_event* sd_future_get_event(sd_future *f) {
+        assert_return(f, NULL);
+        return f->event;
+}
+
 int sd_future_state(sd_future *f) {
         assert_return(f, -EINVAL);
         return f->state;
@@ -166,11 +214,6 @@ int sd_future_result(sd_future *f) {
         return f->result;
 }
 
-void* sd_future_get_userdata(sd_future *f) {
-        assert_return(f, NULL);
-        return f->userdata;
-}
-
 void* sd_future_get_private(sd_future *f) {
         assert_return(f, NULL);
         return f->private;
@@ -181,11 +224,102 @@ const sd_future_ops* sd_future_get_ops(sd_future *f) {
         return f->ops;
 }
 
-int sd_future_set_callback(sd_future *f, sd_future_func_t callback, void *userdata) {
-        assert_return(f, -EINVAL);
+sd_future* sd_future_slot_get_future(sd_future_slot *s) {
+        assert_return(s, NULL);
+        return s->future;
+}
 
-        f->callback = callback;
-        f->userdata = userdata;
+static sd_future_slot* sd_future_slot_free(sd_future_slot *s) {
+        if (!s)
+                return NULL;
+
+        if (s->future) {
+                set_remove(s->future->slots, s);
+                if (!s->floating) {
+                        /* A non-floating slot owns a ref on its future, so this may be the last unref.
+                         * Like any last unref it requires the future to already be RESOLVED — dropping a
+                         * slot must never be what abandons a still-PENDING future. Assert here so the
+                         * misuse points at the slot drop rather than surfacing as the generic
+                         * sd_future_free() assert. */
+                        assert(s->future->n_ref > 1 || s->future->state == SD_FUTURE_RESOLVED);
+                        sd_future_unref(s->future);
+                }
+        }
+
+        sd_event_source_disable_unref(s->defer_source);
+        sd_event_source_disable_unref(s->exit_source);
+
+        return mfree(s);
+}
+
+DEFINE_TRIVIAL_REF_UNREF_FUNC(sd_future_slot, sd_future_slot, sd_future_slot_free);
+
+static int slot_source_setup(sd_event_source *s) {
+        int r;
+
+        r = sd_event_source_set_enabled(s, SD_EVENT_OFF);
+        if (r < 0)
+                return r;
+
+        return event_source_inherit_fiber_priority(s);
+}
+
+int sd_future_add_callback(sd_future *f, sd_future_slot **ret_slot, sd_future_func_t callback, void *userdata) {
+        int r;
+
+        assert_return(f, -EINVAL);
+        assert_return(callback, -EINVAL);
+
+        _cleanup_(sd_future_slot_unrefp) sd_future_slot *s = new(sd_future_slot, 1);
+        if (!s)
+                return -ENOMEM;
+
+        *s = (sd_future_slot) {
+                .n_ref = 1,
+                .future = f,
+                .floating = ret_slot == NULL,
+                .callback = callback,
+                .userdata = userdata,
+        };
+
+        /* Asymmetric ownership (same trick as sd_bus_slot): non-floating slot owns the
+         * future, floating slot is owned by it — avoids a cycle in either direction. */
+        if (!s->floating)
+                sd_future_ref(f);
+
+        /* Never run the callback inline, even if the future is already resolved. Queue it for either
+         * regular dispatch or exit, so callers need not handle synchronous or missed notifications. */
+
+        r = sd_event_add_defer(f->event, &s->defer_source, slot_dispatch_handler, s);
+        if (r < 0)
+                return r;
+
+        r = slot_source_setup(s->defer_source);
+        if (r < 0)
+                return r;
+
+        r = sd_event_add_exit(f->event, &s->exit_source, slot_dispatch_handler, s);
+        if (r < 0)
+                return r;
+
+        r = slot_source_setup(s->exit_source);
+        if (r < 0)
+                return r;
+
+        if (f->state == SD_FUTURE_RESOLVED) {
+                r = slot_arm(s);
+                if (r < 0)
+                        return r;
+        }
+
+        r = set_ensure_put(&f->slots, &trivial_hash_ops, s);
+        if (r < 0)
+                return r;
+
+        if (ret_slot)
+                *ret_slot = s;
+
+        TAKE_PTR(s);
         return 0;
 }
 
@@ -199,65 +333,9 @@ int sd_future_set_priority(sd_future *f, int64_t priority) {
 
 int sd_future_cancel(sd_future *f) {
         assert_return(f, -EINVAL);
-        assert_return(f->ops->cancel, -EOPNOTSUPP);
 
         if (f->state == SD_FUTURE_RESOLVED)
                 return 0;
 
         return f->ops->cancel(f);
 }
-
-typedef struct WaitFuture {
-        sd_future *target;
-} WaitFuture;
-
-static void* wait_future_alloc(void) {
-        return new0(WaitFuture, 1);
-}
-
-static void wait_future_free(sd_future *f) {
-        WaitFuture *wf = ASSERT_PTR(sd_future_get_private(ASSERT_PTR(f)));
-
-        set_remove(wf->target->waiters, f);
-        sd_future_unref(wf->target);
-        free(wf);
-}
-
-static int wait_future_cancel(sd_future *f) {
-        WaitFuture *wf = ASSERT_PTR(sd_future_get_private(ASSERT_PTR(f)));
-
-        set_remove(wf->target->waiters, f);
-        return sd_future_resolve(f, -ECANCELED);
-}
-
-static const sd_future_ops wait_future_ops = {
-        .size = sizeof(sd_future_ops),
-        .alloc = wait_future_alloc,
-        .free = wait_future_free,
-        .cancel = wait_future_cancel,
-};
-
-int sd_future_new_wait(sd_future *target, sd_future **ret) {
-        int r;
-
-        assert_return(target, -EINVAL);
-        assert_return(ret, -EINVAL);
-
-        _cleanup_(sd_future_unrefp) sd_future *f = NULL;
-        r = sd_future_new(&wait_future_ops, &f);
-        if (r < 0)
-                return r;
-
-        WaitFuture *wf = sd_future_get_private(f);
-        wf->target = sd_future_ref(target);
-
-        if (target->state == SD_FUTURE_RESOLVED)
-                r = sd_future_resolve(f, target->result);
-        else
-                r = set_ensure_put(&target->waiters, &trivial_hash_ops, f);
-        if (r < 0)
-                return r;
-
-        *ret = TAKE_PTR(f);
-        return 0;
-}

src/libsystemd/sd-future/test-fiber-io.c

@@ -1111,6 +1111,48 @@ TEST(fiber_poll_timeout) {
         ASSERT_OK(sd_future_result(f));
 }
 
+typedef struct PollInterruptState {
+        int fd;
+        bool timeout;
+} PollInterruptState;
+
+static int poll_interrupt_fiber(void *userdata) {
+        PollInterruptState *s = ASSERT_PTR(userdata);
+        struct pollfd pfd = { .fd = s->fd, .events = POLLIN };
+
+        {
+                SD_FIBER_TIMEOUT(s->timeout ? 0 : USEC_INFINITY);
+                ASSERT_EQ(sd_fiber_ppoll(&pfd, 1, /* timeout= */ NULL, /* sigmask= */ NULL),
+                          s->timeout ? -ETIME : -ECANCELED);
+                ASSERT_EQ(pfd.revents, 0);
+        }
+
+        ASSERT_OK_ZERO(sd_fiber_yield());
+        return 0;
+}
+
+TEST(fiber_poll_interrupted) {
+        bool timeout;
+
+        FOREACH_ARGUMENT(timeout, false, true) {
+                _cleanup_(sd_event_unrefp) sd_event *e = NULL;
+                _cleanup_(sd_future_unrefp) sd_future *f = NULL;
+                _cleanup_close_pair_ int pipefd[2] = EBADF_PAIR;
+
+                ASSERT_OK_ERRNO(pipe2(pipefd, O_CLOEXEC | O_NONBLOCK));
+                ASSERT_OK(sd_event_new(&e));
+                ASSERT_OK(sd_event_set_exit_on_idle(e, true));
+                PollInterruptState s = { .fd = pipefd[0], .timeout = timeout };
+                ASSERT_OK(sd_fiber_new(e, "poll-interrupt", poll_interrupt_fiber, &s,
+                                       /* destroy= */ NULL, &f));
+                ASSERT_OK_POSITIVE(sd_event_run(e, 0));
+                if (!timeout)
+                        ASSERT_OK(sd_future_cancel(f));
+                ASSERT_OK(sd_event_loop(e));
+                ASSERT_OK_ZERO(sd_future_result(f));
+        }
+}
+
 /* Test: poll with zero timeout (should not block) */
 static int poll_zero_timeout_fiber(void *userdata) {
         int *pipefd = userdata;
@@ -1192,6 +1234,30 @@ TEST(fiber_poll_zero_fds_no_timeout) {
         ASSERT_ERROR(sd_future_result(f), EINVAL);
 }
 
+static int poll_all_entries_skipped_fiber(void *userdata) {
+        int *pipefd = ASSERT_PTR(userdata);
+        struct pollfd negative = { .fd = -EBADF, .events = POLLIN };
+        struct pollfd no_events = { .fd = pipefd[0] };
+
+        ASSERT_ERROR(sd_fiber_ppoll(&negative, 1, /* timeout= */ NULL, /* sigmask= */ NULL), EINVAL);
+        ASSERT_ERROR(sd_fiber_ppoll(&no_events, 1, /* timeout= */ NULL, /* sigmask= */ NULL), EINVAL);
+        return 0;
+}
+
+TEST(fiber_poll_all_entries_skipped) {
+        _cleanup_(sd_event_unrefp) sd_event *e = NULL;
+        _cleanup_(sd_future_unrefp) sd_future *f = NULL;
+        _cleanup_close_pair_ int pipefd[2] = EBADF_PAIR;
+
+        ASSERT_OK_ERRNO(pipe2(pipefd, O_NONBLOCK | O_CLOEXEC));
+        ASSERT_OK(sd_event_new(&e));
+        ASSERT_OK(sd_event_set_exit_on_idle(e, true));
+        ASSERT_OK(sd_fiber_new(e, "poll-all-skipped", poll_all_entries_skipped_fiber, pipefd,
+                               /* destroy= */ NULL, &f));
+        ASSERT_OK(sd_event_loop(e));
+        ASSERT_OK_ZERO(sd_future_result(f));
+}
+
 /* Test: poll with negative fd (should be ignored) */
 static int poll_negative_fd_fiber(void *userdata) {
         int *pipefd = userdata;

src/libsystemd/sd-varlink/sd-varlink.c

@@ -1164,7 +1164,7 @@ static int varlink_dispatch_fiber(sd_varlink *v, const char *method, sd_varlink_
                 .callback = callback,
         };
 
-        _cleanup_(sd_future_unrefp) sd_future *f = NULL;
+        _cleanup_(sd_future_cancel_unrefp) sd_future *f = NULL;
         r = sd_fiber_new(v->server->event, method, varlink_fiber_entry, d, varlink_fiber_data_destroy, &f);
         if (r < 0)
                 return r;
@@ -1189,6 +1189,8 @@ static int varlink_dispatch_fiber(sd_varlink *v, const char *method, sd_varlink_
         if (r < 0)
                 return r;
 
+        /* Floating self-ref keeps the fiber alive; release our local ref without cancelling. */
+        f = sd_future_unref(f);
         return 0;
 }
 

src/shared/qmp-client.c

@@ -908,9 +908,10 @@ int qmp_client_call_future(
         assert(c);
         assert(command);
         assert(ret);
+        assert_return(qmp_client_get_event(c), -ENOPKG);
 
-        _cleanup_(sd_future_unrefp) sd_future *f = NULL;
-        r = sd_future_new(&qmp_call_future_ops, &f);
+        _cleanup_(sd_future_cancel_unrefp) sd_future *f = NULL;
+        r = sd_future_new(qmp_client_get_event(c), &qmp_call_future_ops, &f);
         if (r < 0)
                 return r;
 
@@ -976,7 +977,7 @@ static int qmp_client_call_suspend(
         if (r < 0)
                 return r;
 
-        r = sd_fiber_suspend();
+        r = sd_fiber_await(call);
 
         /* If the future isn't resolved, the suspend was interrupted before a reply arrived (fiber
          * cancelled, fiber-wide SD_FIBER_TIMEOUT scope expired, …). There's no reply to extract,

src/systemd/sd-future.h

@@ -32,7 +32,8 @@ struct timespec;
 typedef struct sd_event sd_event;
 typedef struct sd_future sd_future;
 typedef struct sd_future_ops sd_future_ops;
-typedef int (*sd_future_func_t)(sd_future *f);
+typedef struct sd_future_slot sd_future_slot;
+typedef int (*sd_future_func_t)(sd_future *f, void *userdata);
 typedef int (*sd_fiber_func_t)(void *userdata);
 typedef _sd_destroy_t sd_fiber_destroy_t;
 
@@ -50,15 +51,27 @@ __extension__ typedef enum _SD_ENUM_TYPE_S64(sd_future_state_t) {
         _SD_ENUM_FORCE_S64(SD_FUTURE_STATE)
 } sd_future_state_t;
 
-int sd_future_new(const sd_future_ops *ops, sd_future **ret);
+/* SD_EVENT_DEFAULT selects the calling thread's existing default event loop, or fails with -ENOPKG. */
+int sd_future_new(sd_event *e, const sd_future_ops *ops, sd_future **ret);
 int sd_future_cancel(sd_future *f);
 int sd_future_resolve(sd_future *f, int result);
 
+/* A future must be RESOLVED before its last reference is released; dropping the last reference to a
+ * PENDING future is a programming error that aborts the process. Use sd_future_cancel_unref() when
+ * cancellation resolves synchronously, or sd_future_cancel_wait_unref() from a fiber when it must
+ * wait for resolution. Plain sd_future_unref() does not cancel the future. */
 _SD_DECLARE_TRIVIAL_REF_UNREF_FUNC(sd_future);
 _SD_DEFINE_POINTER_CLEANUP_FUNC(sd_future, sd_future_unref);
 void sd_future_unref_array_clear(sd_future *array[], size_t n);
 void sd_future_unref_array(sd_future *array[], size_t n);
 
+sd_future* sd_future_cancel_unref(sd_future *f);
+_SD_DEFINE_POINTER_CLEANUP_FUNC(sd_future, sd_future_cancel_unref);
+void sd_future_cancel_unref_array_clear(sd_future *array[], size_t n);
+void sd_future_cancel_unref_array(sd_future *array[], size_t n);
+
+/* Cancel and release the caller's reference after resolution. The target must not be the calling
+ * fiber. Asynchronous cancellation must be awaited from a fiber on the same event loop. */
 sd_future* sd_future_cancel_wait_unref(sd_future *f);
 _SD_DEFINE_POINTER_CLEANUP_FUNC(sd_future, sd_future_cancel_wait_unref);
 void sd_future_cancel_wait_unref_array_clear(sd_future *array[], size_t n);
@@ -66,14 +79,52 @@ void sd_future_cancel_wait_unref_array(sd_future *array[], size_t n);
 
 int sd_future_state(sd_future *f);
 int sd_future_result(sd_future *f);
-void* sd_future_get_userdata(sd_future *f);
 void* sd_future_get_private(sd_future *f);
 const sd_future_ops* sd_future_get_ops(sd_future *f);
+sd_event* sd_future_get_event(sd_future *f);
+
+int sd_future_add_callback(sd_future *f, sd_future_slot **ret_slot, sd_future_func_t callback, void *userdata);
+
+int sd_future_new_defer(sd_event *e, int result, sd_future **ret);
+
+_SD_DECLARE_TRIVIAL_REF_UNREF_FUNC(sd_future_slot);
+_SD_DEFINE_POINTER_CLEANUP_FUNC(sd_future_slot, sd_future_slot_unref);
+
+sd_future* sd_future_slot_get_future(sd_future_slot *s);
 
-int sd_future_set_callback(sd_future *f, sd_future_func_t callback, void *userdata);
 int sd_future_set_priority(sd_future *f, int64_t priority);
 
-int sd_future_new_wait(sd_future *target, sd_future **ret);
+/* Group policies select an outcome as follows:
+ * WAIT_ALL: first child error, or 0 once every child succeeds.
+ * WAIT_ALL | IGNORE_ERRORS: wait for every child, then return the first error or 0.
+ * WAIT_ANY: a successful child's result if any have succeeded, otherwise the first child error.
+ * WAIT_ANY | IGNORE_ERRORS: wait for a success, or for every child to fail; return that success or
+ *                         the first error, respectively.
+ * "First" means insertion order among the children already resolved when the outcome is selected.
+ * Once selected, the outcome is final: remaining children are cancelled, and the group resolves only
+ * after every child has finished. IGNORE_ERRORS also suppresses cancellation of the parent fiber;
+ * it does not turn an unsuccessful group result into success. Empty groups stay pending until cancelled. */
+__extension__ typedef enum _SD_ENUM_TYPE_S64(sd_future_group_policy_t) {
+        SD_FUTURE_GROUP_WAIT_ALL      = 0,
+        SD_FUTURE_GROUP_WAIT_ANY      = 1 << 0,
+        SD_FUTURE_GROUP_IGNORE_ERRORS = 1 << 1,
+        _SD_FUTURE_GROUP_POLICY_MASK = SD_FUTURE_GROUP_WAIT_ANY | SD_FUTURE_GROUP_IGNORE_ERRORS,
+        _SD_ENUM_FORCE_S64(SD_FUTURE_GROUP_POLICY)
+} sd_future_group_policy_t;
+
+/* The calling fiber becomes the parent only if it belongs to e. Unless IGNORE_ERRORS is set, a child
+ * error cancels the parent if it is not awaiting the group with sd_fiber_await(). Explicit group
+ * cancellation does not cancel the parent. */
+int sd_future_group_new(sd_event *e, sd_future **ret);
+int sd_future_group_set_policy(sd_future *f, uint64_t policy);
+/* The group and all its children must belong to the same event loop. Adding a child takes a new
+ * reference; it does not consume the caller's reference. To hand ownership to the group, release the
+ * caller's reference with sd_future_unref(), not a cancellation cleanup helper. */
+int sd_future_group_add(sd_future *f, sd_future *child);
+int sd_future_group_add_many_internal(sd_future *f, ...) _sd_sentinel_;
+#define sd_future_group_add_many(f, ...) sd_future_group_add_many_internal(f, __VA_ARGS__, NULL)
+/* NULL is treated as an empty group. */
+size_t sd_future_group_size(sd_future *f);
 
 int sd_fiber_new(sd_event *e, const char *name, sd_fiber_func_t func, void *userdata, sd_fiber_destroy_t destroy, sd_future **ret);
 
@@ -84,9 +135,17 @@ int sd_fiber_is_running(void);
 sd_future* sd_fiber_get_current(void);
 int sd_fiber_get_priority(int64_t *ret);
 sd_event* sd_fiber_get_event(void);
+/* The future a fiber is suspended in sd_fiber_await() or sd_future_cancel_wait_unref() for, if any.
+ * The pointer is borrowed and only valid until the fiber runs again. */
+sd_future* sd_fiber_get_awaiting(sd_future *f);
 
 int sd_fiber_yield(void);
 int sd_fiber_sleep(uint64_t usec);
+/* Return the target's result on completion, or a wait error if the calling fiber is interrupted or
+ * the wait fails. An interrupted wait does not imply that the target has resolved; once it has,
+ * sd_future_result() gives its outcome independently of the wait's return value. An already-resolved
+ * target returns its result without consuming a pending interruption. The target must belong to the
+ * calling fiber's event loop. */
 int sd_fiber_await(sd_future *target);
 int sd_fiber_suspend(void);
 int sd_fiber_resume(sd_future *f, int result);

src/test/test-qmp-client.c

@@ -329,6 +329,19 @@ TEST(qmp_call) {
         run_qmp_test(mock_qmp_call_fiber, qmp_client_call_fiber);
 }
 
+TEST(qmp_call_future_without_event) {
+        _cleanup_(qmp_client_unrefp) QmpClient *client = NULL;
+        _cleanup_(sd_future_cancel_unrefp) sd_future *f = NULL;
+        _cleanup_close_pair_ int fds[2] = EBADF_PAIR;
+
+        ASSERT_OK_ERRNO(socketpair(AF_UNIX, SOCK_STREAM|SOCK_CLOEXEC, 0, fds));
+        ASSERT_OK(qmp_client_connect_fd(&client, TAKE_FD(fds[0])));
+        ASSERT_NULL(qmp_client_get_event(client));
+
+        ASSERT_ERROR(ASSERT_RETURN_EXPECTED(qmp_client_call_future(client, "query-status", /* args= */ NULL, &f)), ENOPKG);
+        ASSERT_NULL(f);
+}
+
 static int mock_qmp_call_disconnect_fiber(void *userdata) {
         _cleanup_(json_stream_done) JsonStream s = {};
         _cleanup_(sd_json_variant_unrefp) sd_json_variant *stop_cmd = NULL;
@@ -354,6 +367,84 @@ TEST(qmp_call_disconnect) {
         run_qmp_test(mock_qmp_call_disconnect_fiber, qmp_client_call_disconnect_fiber);
 }
 
+typedef struct QmpInterruptedCall {
+        int fd;
+        sd_future *caller;
+        int error;
+} QmpInterruptedCall;
+
+static int mock_qmp_call_interrupted_fiber(void *userdata) {
+        QmpInterruptedCall *c = ASSERT_PTR(userdata);
+        _cleanup_(json_stream_done) JsonStream s = {};
+        _cleanup_(sd_json_variant_unrefp) sd_json_variant *stop_cmd = NULL, *cont_cmd = NULL;
+
+        mock_qmp_init(&s, c->fd);
+        mock_qmp_handshake(&s);
+        mock_qmp_query_status_running(&s);
+
+        sd_json_variant *stop_id = mock_qmp_expect(&s, "stop", &stop_cmd);
+        if (c->error == ECANCELED)
+                ASSERT_OK(sd_future_cancel(c->caller));
+        else if (c->error == EBUSY)
+                ASSERT_OK(sd_fiber_resume(c->caller, 42));
+
+        /* Wait until the client has abandoned stop and issued another call. Delivering the late
+         * reply first must not wake the new wait or touch the cancelled call's freed future. */
+        sd_json_variant *cont_id = mock_qmp_expect(&s, "cont", &cont_cmd);
+        mock_qmp_reply(&s, stop_id, /* reply_data= */ NULL);
+        mock_qmp_reply(&s, cont_id, /* reply_data= */ NULL);
+        return 0;
+}
+
+static int qmp_client_call_interrupted_fiber(void *userdata) {
+        int error = *(int*) ASSERT_PTR(userdata);
+        _cleanup_(qmp_client_unrefp) QmpClient *client = NULL;
+        _cleanup_(sd_future_cancel_wait_unrefp) sd_future *mock = NULL;
+        _cleanup_(sd_json_variant_unrefp) sd_json_variant *result = NULL;
+        _cleanup_free_ char *error_desc = NULL;
+        _cleanup_close_pair_ int fds[2] = EBADF_PAIR;
+
+        ASSERT_OK_ERRNO(socketpair(AF_UNIX, SOCK_STREAM|SOCK_CLOEXEC, 0, fds));
+        ASSERT_OK(qmp_client_connect_fd(&client, TAKE_FD(fds[0])));
+        ASSERT_OK(qmp_client_attach_event(client, sd_fiber_get_event(), SD_EVENT_PRIORITY_NORMAL));
+        QmpInterruptedCall c = { .fd = TAKE_FD(fds[1]), .caller = sd_fiber_get_current(), .error = error };
+        ASSERT_OK(sd_fiber_new(sd_fiber_get_event(), "interrupted-mock", mock_qmp_call_interrupted_fiber,
+                               &c, /* destroy= */ NULL, &mock));
+
+        /* Finish the handshake before testing interruption of a command awaiting its reply. */
+        ASSERT_OK_POSITIVE(qmp_client_call(client, "query-status", /* args= */ NULL,
+                                          /* ret_result= */ NULL, /* reterr_error_desc= */ NULL));
+        {
+                SD_FIBER_TIMEOUT(error == ETIME ? 0 : USEC_INFINITY);
+                ASSERT_ERROR(qmp_client_call(client, "stop", /* args= */ NULL, &result, &error_desc), error);
+        }
+        ASSERT_NULL(result);
+        ASSERT_NULL(error_desc);
+
+        ASSERT_OK_POSITIVE(qmp_client_call(client, "cont", /* args= */ NULL, &result, &error_desc));
+        ASSERT_NOT_NULL(result);
+        ASSERT_NULL(error_desc);
+        ASSERT_OK_ZERO(sd_fiber_await(mock));
+        ASSERT_OK_ZERO(sd_fiber_yield());
+        ASSERT_OK(sd_event_exit(sd_fiber_get_event(), 0));
+        return 0;
+}
+
+TEST(qmp_call_interrupted) {
+        int error;
+
+        FOREACH_ARGUMENT(error, ECANCELED, ETIME, EBUSY) {
+                _cleanup_(sd_event_unrefp) sd_event *e = NULL;
+                _cleanup_(sd_future_unrefp) sd_future *caller = NULL;
+
+                ASSERT_OK(sd_event_new(&e));
+                ASSERT_OK(sd_fiber_new(e, "interrupted-client", qmp_client_call_interrupted_fiber,
+                                       &error, /* destroy= */ NULL, &caller));
+                ASSERT_OK(sd_event_loop(e));
+                ASSERT_OK_ZERO(sd_future_result(caller));
+        }
+}
+
 static int mock_qmp_fd_fiber(void *userdata) {
         _cleanup_(json_stream_done) JsonStream s = {};
         _cleanup_(sd_json_variant_unrefp) sd_json_variant *cap_cmd = NULL, *addfd_cmd = NULL,