repart: support subscription on ListCandidateDevices

Extend the Varlink call so it can optionally stay open after the initial enumeration and stream additions and removals over the same connection. Useful for UIs that want a live list of candidate disks without polling. New input "subscribe" (default false) and new output "action" (an enum "add" / "remove" / "ready"). When subscribing: - The server starts an sd-device monitor on the "block" subsystem *before* the initial enumeration so events occurring during that window aren't dropped. Duplicate "add" events for the same device are legitimate and documented; clients upsert by identifier. - The initial candidate set is streamed as "add" notifications. - Exactly one "ready" notification follows, marking the boundary between the initial picture and live deltas. - Live SD_DEVICE_REMOVE events become "remove" notifications carrying only "node". - All other live actions re-run blockdev_list_one(); MATCH_YES maps to "add" (which doubles as "device metadata changed"), MATCH_FILTERED maps to "remove" so transitions driven by the dynamic ignoreEmpty / ignoreReadOnly filters (size becoming 0, read-only flag flipping on) are propagated as the device leaving the candidate set. Clients may therefore receive a "remove" for a device they never saw an "add" for; that is documented as legal. Existing non-subscribing callers see no schema change: the new "action" field is omitted from their replies via SD_JSON_BUILD_PAIR_CONDITION, and the original NoCandidateDevices sentinel still fires when the enumeration is empty. The per-link subscription state is freed via an on-disconnect callback bound lazily on first subscribe.
2026-06-01 11:26:15 +02:00
parent 5af1038123
commit 81bd798b65
2 changed files with 218 additions and 29 deletions
@@ -1,12 +1,134 @@
 /* SPDX-License-Identifier: LGPL-2.1-or-later */

+#include "sd-device.h"
 #include "sd-json.h"
 #include "sd-varlink.h"

 #include "blockdev-list.h"
+#include "device-util.h"
 #include "json-util.h"
 #include "repart-list-candidate-devices.h"

+typedef struct ListCandidateDevicesContext {
+        sd_varlink *link;                  /* Owned ref; freed in list_candidate_devices_context_free(). */
+        sd_device_monitor *monitor;
+        BlockDevListFlags flags;
+        dev_t root_devno, whole_root_devno;
+} ListCandidateDevicesContext;
+
+static ListCandidateDevicesContext* list_candidate_devices_context_free(ListCandidateDevicesContext *c) {
+        if (!c)
+                return NULL;
+
+        sd_device_monitor_unref(c->monitor);
+        sd_varlink_unref(c->link);
+        return mfree(c);
+}
+
+DEFINE_TRIVIAL_CLEANUP_FUNC(ListCandidateDevicesContext*, list_candidate_devices_context_free);
+
+static void vl_on_disconnect(sd_varlink_server *server, sd_varlink *link, void *userdata) {
+        assert(server);
+        assert(link);
+
+        list_candidate_devices_context_free(sd_varlink_set_userdata(link, NULL));
+}
+
+static int list_candidate_devices_send_remove(sd_varlink *link, sd_device *dev) {
+        int r;
+
+        assert(link);
+        assert(dev);
+
+        const char *node;
+        r = sd_device_get_devname(dev, &node);
+        if (r < 0) {
+                log_device_debug_errno(dev, r, "Failed to acquire device node of removed block device, ignoring: %m");
+                return 0;
+        }
+
+        return sd_varlink_notifybo(link,
+                        SD_JSON_BUILD_PAIR("action", JSON_BUILD_CONST_STRING("remove")),
+                        SD_JSON_BUILD_PAIR_STRING("node", node));
+}
+
+static int list_candidate_devices_send_add(sd_varlink *link, const BlockDevice *d, bool with_action) {
+        int r;
+
+        assert(link);
+        assert(d);
+
+        /* In subscribe mode we tag every reply with action=add so the discriminator is meaningful;
+         * in plain enumeration mode we omit it (older clients don't know the field). */
+        _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL;
+        r = sd_json_buildo(
+                        &v,
+                        SD_JSON_BUILD_PAIR_CONDITION(with_action, "action", JSON_BUILD_CONST_STRING("add")),
+                        SD_JSON_BUILD_PAIR_STRING("node", d->node),
+                        JSON_BUILD_PAIR_STRV_NON_EMPTY("symlinks", d->symlinks),
+                        JSON_BUILD_PAIR_UNSIGNED_NOT_EQUAL("diskseq", d->diskseq, UINT64_MAX),
+                        JSON_BUILD_PAIR_UNSIGNED_NOT_EQUAL("sizeBytes", d->size, UINT64_MAX),
+                        JSON_BUILD_PAIR_STRING_NON_EMPTY("model", d->model),
+                        JSON_BUILD_PAIR_STRING_NON_EMPTY("vendor", d->vendor),
+                        JSON_BUILD_PAIR_STRING_NON_EMPTY("subsystem", d->subsystem));
+        if (r < 0)
+                return r;
+
+        /* Subscribe mode sets no sentinel, so a final reply would terminate the streaming call after
+         * the first device; use intermediate notify messages to keep it open. Plain enumeration relies
+         * on the sentinel mechanism, which buffers and chains replies. */
+        return with_action ? sd_varlink_notify(link, v) : sd_varlink_reply(link, v);
+}
+
+static int list_candidate_devices_on_uevent(sd_device_monitor *monitor, sd_device *dev, void *userdata) {
+        ListCandidateDevicesContext *c = ASSERT_PTR(userdata);
+        int r;
+
+        assert(dev);
+
+        sd_device_action_t action;
+        r = sd_device_get_action(dev, &action);
+        if (r < 0) {
+                log_device_debug_errno(dev, r, "Failed to acquire uevent action of block device, ignoring: %m");
+                return 0;
+        }
+
+        if (action == SD_DEVICE_REMOVE)
+                (void) list_candidate_devices_send_remove(c->link, dev);
+        else {
+                /* All non-REMOVE actions are treated as "device exists afterwards". Re-run the full
+                 * filter set; the tri-state distinguishes "ignore entirely" (static-filtered) from
+                 * "client should treat this as a removal" (dynamic-filtered, e.g. size became 0 or
+                 * read-only became 1). */
+                _cleanup_(block_device_done) BlockDevice d = BLOCK_DEVICE_NULL;
+                r = blockdev_list_one(dev, c->flags, c->root_devno, c->whole_root_devno, &d);
+                if (r < 0)
+                        return r;
+
+                switch (r) {
+                case BLOCKDEV_LIST_MATCH_NO:
+                case BLOCKDEV_LIST_MATCH_SKIPPED:
+                        break;
+
+                case BLOCKDEV_LIST_MATCH_FILTERED:
+                        (void) list_candidate_devices_send_remove(c->link, dev);
+                        break;
+
+                case BLOCKDEV_LIST_MATCH_YES:
+                        (void) list_candidate_devices_send_add(c->link, &d, /* with_action= */ true);
+                        break;
+
+                default:
+                        assert_not_reached();
+                }
+        }
+
+        /* Push out any notifications we just queued so the client doesn't have to wait for the next
+         * event-loop tick to receive them. */
+        (void) sd_varlink_flush(c->link);
+        return 0;
+}
+
 int vl_method_list_candidate_devices(
                sd_varlink *link,
                sd_json_variant *parameters,
@@ -16,11 +138,13 @@ int vl_method_list_candidate_devices(
        struct {
                bool ignore_root;
                bool ignore_empty;
+                bool subscribe;
        } p = {};

        static const sd_json_dispatch_field dispatch_table[] = {
                { "ignoreRoot",  SD_JSON_VARIANT_BOOLEAN, sd_json_dispatch_stdbool, voffsetof(p, ignore_root),  0 },
                { "ignoreEmpty", SD_JSON_VARIANT_BOOLEAN, sd_json_dispatch_stdbool, voffsetof(p, ignore_empty), 0 },
+                { "subscribe",   SD_JSON_VARIANT_BOOLEAN, sd_json_dispatch_stdbool, voffsetof(p, subscribe),    0 },
                {}
        };

@@ -33,39 +157,89 @@ int vl_method_list_candidate_devices(
        if (r != 0)
                return r;

-        BlockDevice *l = NULL;
-        size_t n = 0;
-        CLEANUP_ARRAY(l, n, block_device_array_free);
+        const BlockDevListFlags listflags =
+                BLOCKDEV_LIST_SHOW_SYMLINKS|
+                BLOCKDEV_LIST_REQUIRE_PARTITION_SCANNING|
+                BLOCKDEV_LIST_IGNORE_ZRAM|
+                BLOCKDEV_LIST_METADATA|
+                BLOCKDEV_LIST_IGNORE_READ_ONLY|
+                (p.ignore_empty ? BLOCKDEV_LIST_IGNORE_EMPTY : 0)|
+                (p.ignore_root ? BLOCKDEV_LIST_IGNORE_ROOT : 0);

-        r = blockdev_list(
-                        BLOCKDEV_LIST_SHOW_SYMLINKS|
-                        BLOCKDEV_LIST_REQUIRE_PARTITION_SCANNING|
-                        BLOCKDEV_LIST_IGNORE_ZRAM|
-                        BLOCKDEV_LIST_METADATA|
-                        BLOCKDEV_LIST_IGNORE_READ_ONLY|
-                        (p.ignore_empty ? BLOCKDEV_LIST_IGNORE_EMPTY : 0)|
-                        (p.ignore_root ? BLOCKDEV_LIST_IGNORE_ROOT : 0),
-                        &l,
-                        &n);
-        if (r < 0)
-                return r;
+        _cleanup_(list_candidate_devices_context_freep) ListCandidateDevicesContext *c = NULL;

-        r = sd_varlink_set_sentinel(link, "io.systemd.Repart.NoCandidateDevices");
-        if (r < 0)
-                return r;
+        if (p.subscribe) {
+                /* Start the monitor *before* the initial enumeration so we don't lose events that fire
+                 * during the enumeration window. Duplicate add events for the same device are
+                 * legitimate and documented; clients upsert by identifier. */

-        FOREACH_ARRAY(d, l, n) {
-                r = sd_varlink_replybo(link,
-                                SD_JSON_BUILD_PAIR_STRING("node", d->node),
-                                JSON_BUILD_PAIR_STRV_NON_EMPTY("symlinks", d->symlinks),
-                                JSON_BUILD_PAIR_UNSIGNED_NOT_EQUAL("diskseq", d->diskseq, UINT64_MAX),
-                                JSON_BUILD_PAIR_UNSIGNED_NOT_EQUAL("sizeBytes", d->size, UINT64_MAX),
-                                JSON_BUILD_PAIR_STRING_NON_EMPTY("model", d->model),
-                                JSON_BUILD_PAIR_STRING_NON_EMPTY("vendor", d->vendor),
-                                JSON_BUILD_PAIR_STRING_NON_EMPTY("subsystem", d->subsystem));
+                c = new(ListCandidateDevicesContext, 1);
+                if (!c)
+                        return -ENOMEM;
+
+                *c = (ListCandidateDevicesContext) {
+                        .flags = listflags,
+                };
+
+                (void) blockdev_list_get_root_devnos(listflags, &c->root_devno, &c->whole_root_devno);
+
+                r = sd_device_monitor_new(&c->monitor);
+                if (r < 0)
+                        return log_error_errno(r, "Failed to allocate block device monitor: %m");
+
+                (void) sd_device_monitor_set_description(c->monitor, "repart-candidate-devices");
+
+                r = sd_device_monitor_filter_add_match_subsystem_devtype(c->monitor, "block", /* devtype= */ NULL);
+                if (r < 0)
+                        return log_error_errno(r, "Failed to configure block device monitor filter: %m");
+
+                r = sd_device_monitor_attach_event(c->monitor, sd_varlink_get_event(link));
+                if (r < 0)
+                        return log_error_errno(r, "Failed to attach block device monitor to event loop: %m");
+
+                r = sd_device_monitor_start(c->monitor, list_candidate_devices_on_uevent, c);
+                if (r < 0)
+                        return log_error_errno(r, "Failed to start block device monitor: %m");
+        } else {
+                /* Non-subscribing: the sentinel fires if no replies were sent. */
+                r = sd_varlink_set_sentinel(link, "io.systemd.Repart.NoCandidateDevices");
                if (r < 0)
                        return r;
        }

+        BlockDevice *l = NULL;
+        size_t n = 0;
+        CLEANUP_ARRAY(l, n, block_device_array_free);
+
+        r = blockdev_list(listflags, &l, &n);
+        if (r < 0)
+                return r;
+
+        FOREACH_ARRAY(d, l, n) {
+                r = list_candidate_devices_send_add(link, d, /* with_action= */ p.subscribe);
+                if (r < 0)
+                        return r;
+        }
+
+        if (!p.subscribe)
+                return 0;
+
+        /* Subscribing: a single "ready" sentinel marks the boundary between initial enumeration and
+         * live events. The call stays open after we return; live events flow through
+         * list_candidate_devices_on_uevent(). */
+        r = sd_varlink_notifybo(link, SD_JSON_BUILD_PAIR("action", JSON_BUILD_CONST_STRING("ready")));
+        if (r < 0)
+                return r;
+
+        /* Install the on-disconnect handler lazily, only once we actually have a subscription to clean
+         * up. The call is idempotent for the same callback. */
+        r = sd_varlink_server_bind_disconnect(sd_varlink_get_server(link), vl_on_disconnect);
+        if (r < 0)
+                return r;
+
+        /* Hand ownership of the context to the link; vl_on_disconnect() frees it. */
+        c->link = sd_varlink_ref(link);
+        sd_varlink_set_userdata(link, TAKE_PTR(c));
+
        return 0;
 }
@@ -29,6 +29,15 @@ static SD_VARLINK_DEFINE_ENUM_TYPE(
                SD_VARLINK_FIELD_COMMENT("Always create a new partition table, potentially overwriting an existing table"),
                SD_VARLINK_DEFINE_ENUM_VALUE(force));

+static SD_VARLINK_DEFINE_ENUM_TYPE(
+                BlockDeviceAction,
+                SD_VARLINK_FIELD_COMMENT("The device is currently present and a candidate. Emitted both during the initial enumeration and for live uevents (any action other than 'remove' is propagated as 'add', including the synthesized transitions when a device becomes empty or read-only and a relevant ignore* input is in effect)."),
+                SD_VARLINK_DEFINE_ENUM_VALUE(add),
+                SD_VARLINK_FIELD_COMMENT("The device is no longer a candidate, either because the kernel reported a remove event, or because a dynamic filter (ignoreEmpty/ignoreReadOnly) flipped it out of the candidate set. Clients should drop the device identified by 'node'. It is permitted (and harmless) to receive a 'remove' for a device the client never saw an 'add' for; clients should ignore these silently."),
+                SD_VARLINK_DEFINE_ENUM_VALUE(remove),
+                SD_VARLINK_FIELD_COMMENT("Sentinel sent exactly once in subscribe mode, after the initial enumeration is complete. Everything that follows reflects live uevents. Carries no other fields."),
+                SD_VARLINK_DEFINE_ENUM_VALUE(ready));
+
 static SD_VARLINK_DEFINE_METHOD_FULL(
                Run,
                SD_VARLINK_SUPPORTS_MORE,
@@ -64,8 +73,12 @@ static SD_VARLINK_DEFINE_METHOD_FULL(
                SD_VARLINK_DEFINE_INPUT(ignoreRoot, SD_VARLINK_BOOL, SD_VARLINK_NULLABLE),
                SD_VARLINK_FIELD_COMMENT("Control whether to include block devices with zero size in the list, i.e. typically block devices without any inserted medium. Defaults to false, i.e. empty block devices are included."),
                SD_VARLINK_DEFINE_INPUT(ignoreEmpty, SD_VARLINK_BOOL, SD_VARLINK_NULLABLE),
-                SD_VARLINK_FIELD_COMMENT("The device node path of the block device."),
-                SD_VARLINK_DEFINE_OUTPUT(node, SD_VARLINK_STRING, 0),
+                SD_VARLINK_FIELD_COMMENT("If true, keep the call open after the initial enumeration and stream live add/remove notifications as block-subsystem uevents arrive. The end of the initial enumeration is marked by exactly one notification with action='ready' and no other fields. Defaults to false."),
+                SD_VARLINK_DEFINE_INPUT(subscribe, SD_VARLINK_BOOL, SD_VARLINK_NULLABLE),
+                SD_VARLINK_FIELD_COMMENT("Discriminator field. Only set in subscribe mode. 'add' carries the full device record, 'remove' carries only node, 'ready' carries no other fields and is sent once after the initial enumeration."),
+                SD_VARLINK_DEFINE_OUTPUT_BY_TYPE(action, BlockDeviceAction, SD_VARLINK_NULLABLE),
+                SD_VARLINK_FIELD_COMMENT("The device node path of the block device. Identifies the device on 'remove' too."),
+                SD_VARLINK_DEFINE_OUTPUT(node, SD_VARLINK_STRING, SD_VARLINK_NULLABLE),
                SD_VARLINK_FIELD_COMMENT("List of symlinks pointing to the device node, if any."),
                SD_VARLINK_DEFINE_OUTPUT(symlinks, SD_VARLINK_STRING, SD_VARLINK_ARRAY|SD_VARLINK_NULLABLE),
                SD_VARLINK_FIELD_COMMENT("The Linux kernel disk sequence number identifying the medium."),
@@ -106,6 +119,8 @@ SD_VARLINK_DEFINE_INTERFACE(
                &vl_type_EmptyMode,
                SD_VARLINK_SYMBOL_COMMENT("Progress phase identifiers. Note that we might add more phases here, and thus identifiers. Frontends can choose to display the phase to the user in some human readable form, or not do that, but if they do it and they receive a notification for a so far unknown phase, they should just ignore it."),
                &vl_type_ProgressPhase,
+                SD_VARLINK_SYMBOL_COMMENT("Discriminator on streamed ListCandidateDevices replies in subscribe mode."),
+                &vl_type_BlockDeviceAction,

                SD_VARLINK_SYMBOL_COMMENT("Invoke the actual repartitioning operation, either in dry-run mode or for real. If invoked with 'more' enabled will report progress, otherwise will just report completion."),
                &vl_method_Run,