manager: add EnqueueMarkedJobs varlink method

Same as the D-Bus method. Returns array of job IDs.

src/core/dbus-unit.c

@@ -24,9 +24,9 @@
 #include "manager.h"
 #include "namespace-util.h"
 #include "path-util.h"
+#include "pidref.h"
 #include "process-util.h"
 #include "selinux-access.h"
-#include "service.h"
 #include "set.h"
 #include "signal-util.h"
 #include "special.h"
@@ -1960,44 +1960,23 @@ int bus_unit_queue_job_one(
         Job *j, *a;
         int r;
 
-        if (FLAGS_SET(flags, BUS_UNIT_QUEUE_RELOAD_IF_POSSIBLE) && unit_can_reload(u)) {
-                if (type == JOB_RESTART)
-                        type = JOB_RELOAD_OR_START;
-                else if (type == JOB_TRY_RESTART)
-                        type = JOB_TRY_RELOAD;
-        }
+        assert(u);
 
-        if (type == JOB_STOP && UNIT_IS_LOAD_ERROR(u->load_state) && unit_active_state(u) == UNIT_INACTIVE)
+        r = unit_queue_job_check_and_collapse_type(u, &type, /* reload_if_possible= */ FLAGS_SET(flags, BUS_UNIT_QUEUE_RELOAD_IF_POSSIBLE));
+        if (r == -ENOENT)
                 return sd_bus_error_setf(reterr_error, BUS_ERROR_NO_SUCH_UNIT, "Unit %s not loaded.", u->id);
-
-        if ((type == JOB_START && u->refuse_manual_start) ||
-            (type == JOB_STOP && u->refuse_manual_stop) ||
-            (IN_SET(type, JOB_RESTART, JOB_TRY_RESTART) && (u->refuse_manual_start || u->refuse_manual_stop)) ||
-            (type == JOB_RELOAD_OR_START && job_type_collapse(type, u) == JOB_START && u->refuse_manual_start))
+        if (r == -EUNATCH)
                 return sd_bus_error_setf(reterr_error,
                                          BUS_ERROR_ONLY_BY_DEPENDENCY,
                                          "Operation refused, unit %s may be requested by dependency only (it is configured to refuse manual start/stop).",
                                          u->id);
-
-        /* dbus-broker issues StartUnit for activation requests, and Type=dbus services automatically
-         * gain dependency on dbus.socket. Therefore, if dbus has a pending stop job, the new start
-         * job that pulls in dbus again would cause job type conflict. Let's avoid that by rejecting
-         * job enqueuing early.
-         *
-         * Note that unlike signal_activation_request(), we can't use unit_inactive_or_pending()
-         * here. StartUnit is a more generic interface, and thus users are allowed to use e.g. systemctl
-         * to start Type=dbus services even when dbus is inactive. */
-        if (type == JOB_START && u->type == UNIT_SERVICE && SERVICE(u)->type == SERVICE_DBUS)
-                FOREACH_STRING(dbus_unit, SPECIAL_DBUS_SOCKET, SPECIAL_DBUS_SERVICE) {
-                        Unit *dbus;
-
-                        dbus = manager_get_unit(u->manager, dbus_unit);
-                        if (dbus && unit_stop_pending(dbus))
-                                return sd_bus_error_setf(reterr_error,
-                                                         BUS_ERROR_SHUTTING_DOWN,
-                                                         "Operation for unit %s refused, D-Bus is shutting down.",
-                                                         u->id);
-                }
+        if (r == -ESHUTDOWN)
+                return sd_bus_error_setf(reterr_error,
+                                         BUS_ERROR_SHUTTING_DOWN,
+                                         "Operation for unit %s refused, D-Bus is shutting down.",
+                                         u->id);
+        if (r < 0)
+                return r;
 
         if (FLAGS_SET(flags, BUS_UNIT_QUEUE_VERBOSE_REPLY)) {
                 affected = set_new(NULL);

src/core/unit.c

@@ -7003,3 +7003,57 @@ UnitDependency unit_mount_dependency_type_to_dependency_type(UnitMountDependency
                 assert_not_reached();
         }
 }
+
+int unit_queue_job_check_and_collapse_type(
+                Unit *u,
+                JobType *type, /* input and output */
+                bool reload_if_possible) {
+
+        /* Returns:
+         *
+         * -ENOENT    → Unit not loaded
+         * -EUNATCH   → Unit can only be activated via dependency, not directly
+         * -ESHUTDOWN → System bus is shutting down */
+
+        JobType t;
+
+        assert(u);
+        assert(type);
+
+        t = *type;
+
+        if (reload_if_possible && unit_can_reload(u)) {
+                if (t == JOB_RESTART)
+                        t = JOB_RELOAD_OR_START;
+                else if (t == JOB_TRY_RESTART)
+                        t = JOB_TRY_RELOAD;
+        }
+
+        if (t == JOB_STOP && UNIT_IS_LOAD_ERROR(u->load_state) && unit_active_state(u) == UNIT_INACTIVE)
+                return -ENOENT;
+
+        if ((t == JOB_START && u->refuse_manual_start) ||
+            (t == JOB_STOP && u->refuse_manual_stop) ||
+            (IN_SET(t, JOB_RESTART, JOB_TRY_RESTART) && (u->refuse_manual_start || u->refuse_manual_stop)) ||
+            (t == JOB_RELOAD_OR_START && job_type_collapse(t, u) == JOB_START && u->refuse_manual_start))
+                return -EUNATCH;
+
+        /* dbus-broker issues StartUnit for activation requests, and Type=dbus services automatically
+         * gain dependency on dbus.socket. Therefore, if dbus has a pending stop job, the new start
+         * job that pulls in dbus again would cause job type conflict. Let's avoid that by rejecting
+         * job enqueuing early.
+         *
+         * Note that unlike signal_activation_request(), we can't use unit_inactive_or_pending()
+         * here. StartUnit is a more generic interface, and thus users are allowed to use e.g. systemctl
+         * to start Type=dbus services even when dbus is inactive. */
+        if (t == JOB_START && u->type == UNIT_SERVICE && SERVICE(u)->type == SERVICE_DBUS)
+                FOREACH_STRING(dbus_unit, SPECIAL_DBUS_SOCKET, SPECIAL_DBUS_SERVICE) {
+                        Unit *dbus = manager_get_unit(u->manager, dbus_unit);
+                        if (dbus && unit_stop_pending(dbus))
+                                return -ESHUTDOWN;
+                }
+
+        *type = t;
+
+        return 0;
+}

src/core/unit.h

@@ -1092,6 +1092,8 @@ UnitDependency unit_mount_dependency_type_to_dependency_type(UnitMountDependency
 
 DECLARE_STRING_TABLE_LOOKUP(oom_policy, OOMPolicy);
 
+int unit_queue_job_check_and_collapse_type(Unit *u, JobType *type, bool reload_if_possible);
+
 /* Macros which append UNIT= or USER_UNIT= to the message */
 
 #define log_unit_full_errno_zerook(unit, level, error, ...)             \

src/core/varlink-manager.c

@@ -6,9 +6,13 @@
 
 #include "alloc-util.h"
 #include "architecture.h"
+#include "bitfield.h"
 #include "build.h"
 #include "bus-polkit.h"
 #include "confidential-virt.h"
+#include "dbus-job.h"
+#include "errno-util.h"
+#include "glyph-util.h"
 #include "json-util.h"
 #include "manager.h"
 #include "pidref.h"
@@ -20,6 +24,7 @@
 #include "version.h"
 #include "varlink-common.h"
 #include "varlink-manager.h"
+#include "varlink-unit.h"
 #include "varlink-util.h"
 #include "virt.h"
 #include "watchdog.h"
@@ -299,3 +304,109 @@ int vl_method_reexecute_manager(sd_varlink *link, sd_json_variant *parameters, s
 
         return 1;
 }
+static int varlink_manager_queue_job_one(
+                sd_varlink *link,
+                Unit *u,
+                JobType type,
+                JobMode mode,
+                uint32_t *ret_job_id) {
+
+        int r;
+
+        assert(u);
+
+        r = unit_queue_job_check_and_collapse_type(u, &type, /* reload_if_possible= */ BIT_SET(u->markers, UNIT_MARKER_NEEDS_RELOAD));
+        if (r == -ENOENT)
+                return varlink_error_no_such_unit(link, "name");
+        if (r == -EUNATCH)
+                return sd_varlink_errorb(link, VARLINK_ERROR_MANAGER_ONLY_BY_DEPENDENCY);
+        if (r == -ESHUTDOWN)
+                return sd_varlink_errorb(link, VARLINK_ERROR_MANAGER_BUS_SHUTTING_DOWN);
+        if (r < 0)
+                return r;
+
+        Job *j;
+        r = manager_add_job(u->manager, type, u, mode, /* reterr_error= */ NULL, &j);
+        if (r < 0)
+                return r;
+
+        /* Before we send the method reply, force out the announcement JobNew for this job */
+        bus_job_send_pending_change_signal(j, /* including_new= */ true);
+
+        if (ret_job_id)
+                *ret_job_id = j->id;
+
+        return 0;
+}
+
+int vl_method_enqueue_marked_jobs_manager(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata) {
+        Manager *manager = ASSERT_PTR(userdata);
+        int r;
+
+        assert(link);
+        assert(parameters);
+
+        r = sd_varlink_dispatch(link, parameters, /* dispatch_table= */ NULL, /* userdata= */ NULL);
+        if (r != 0)
+                return r;
+
+        r = mac_selinux_access_check_varlink(link, "start");
+        if (r < 0)
+                return r;
+
+        r = varlink_verify_polkit_async(
+                        link,
+                        manager->system_bus,
+                        "org.freedesktop.systemd1.manage-units",
+                        /* details= */ NULL,
+                        &manager->polkit_registry);
+        if (r <= 0)
+                return r;
+
+        log_info("Queuing reload/restart jobs for marked units%s", glyph(GLYPH_ELLIPSIS));
+
+        _cleanup_(sd_json_variant_unrefp) sd_json_variant *array = NULL, *reply = NULL;
+        Unit *u;
+        char *k;
+        int ret = 0;
+        HASHMAP_FOREACH_KEY(u, k, manager->units) {
+                uint32_t job_id = 0; /* silence 'maybe-uninitialized' compiler warning */
+
+                /* ignore aliases */
+                if (u->id != k)
+                        continue;
+
+                if (u->markers == 0)
+                        continue;
+
+                r = mac_selinux_unit_access_check_varlink(u, link, job_type_to_access_method(JOB_TRY_RESTART));
+                if (r >= 0)
+                        r = varlink_manager_queue_job_one(
+                                        link,
+                                        u,
+                                        JOB_TRY_RESTART,
+                                        JOB_FAIL,
+                                        &job_id);
+                if (ERRNO_IS_NEG_RESOURCE(r))
+                        return r;
+                RET_GATHER(ret, r);
+                if (r >= 0) {
+                        r = sd_json_variant_append_arrayb(&array, SD_JSON_BUILD_UNSIGNED(job_id));
+                        if (r < 0)
+                                return r;
+                }
+        }
+
+        if (ret < 0)
+                return ret;
+
+        /* Return parameter is not nullable, build empty array if there's nothing to return */
+        if (array)
+                r = sd_json_buildo(&reply, SD_JSON_BUILD_PAIR_VARIANT("JobIDs", array));
+        else
+                r = sd_json_buildo(&reply, SD_JSON_BUILD_PAIR_EMPTY_ARRAY("JobIDs"));
+        if (r < 0)
+                return r;
+
+        return sd_varlink_reply(link, reply);
+}

src/core/varlink-manager.h

@@ -4,7 +4,10 @@
 #include "core-forward.h"
 
 #define VARLINK_ERROR_MANAGER_RATE_LIMIT_REACHED "io.systemd.Manager.RateLimitReached"
+#define VARLINK_ERROR_MANAGER_ONLY_BY_DEPENDENCY "io.systemd.Manager.OnlyByDependency"
+#define VARLINK_ERROR_MANAGER_BUS_SHUTTING_DOWN "io.systemd.Manager.BusShuttingDown"
 
 int vl_method_describe_manager(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata);
 int vl_method_reexecute_manager(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata);
 int vl_method_reload_manager(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata);
+int vl_method_enqueue_marked_jobs_manager(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata);

src/core/varlink-unit.c

@@ -382,7 +382,7 @@ static void unit_lookup_parameters_done(UnitLookupParameters *p) {
         pidref_done(&p->pidref);
 }
 
-static int varlink_error_no_such_unit(sd_varlink *v, const char *name) {
+int varlink_error_no_such_unit(sd_varlink *v, const char *name) {
         return sd_varlink_errorbo(
                         ASSERT_PTR(v),
                         VARLINK_ERROR_UNIT_NO_SUCH_UNIT,

src/core/varlink-unit.h

@@ -5,4 +5,5 @@
 
 #define VARLINK_ERROR_UNIT_NO_SUCH_UNIT "io.systemd.Unit.NoSuchUnit"
 
+int varlink_error_no_such_unit(sd_varlink *v, const char *name);
 int vl_method_list_units(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata);

src/core/varlink.c

@@ -386,6 +386,7 @@ int manager_setup_varlink_server(Manager *m) {
                         "io.systemd.Manager.Describe", vl_method_describe_manager,
                         "io.systemd.Manager.Reexecute", vl_method_reexecute_manager,
                         "io.systemd.Manager.Reload", vl_method_reload_manager,
+                        "io.systemd.Manager.EnqueueMarkedJobs", vl_method_enqueue_marked_jobs_manager,
                         "io.systemd.Unit.List", vl_method_list_units,
                         "io.systemd.service.Ping", varlink_method_ping,
                         "io.systemd.service.GetEnvironment", varlink_method_get_environment);

src/shared/varlink-io.systemd.Manager.c

@@ -181,8 +181,17 @@ static SD_VARLINK_DEFINE_METHOD(
 static SD_VARLINK_DEFINE_METHOD(
                 Reload);
 
+static SD_VARLINK_DEFINE_METHOD(
+                EnqueueMarkedJobs,
+                SD_VARLINK_FIELD_COMMENT("IDs of enqueued jobs"),
+                SD_VARLINK_DEFINE_OUTPUT(JobIDs, SD_VARLINK_INT, SD_VARLINK_ARRAY));
+
 static SD_VARLINK_DEFINE_ERROR(RateLimitReached);
 
+static SD_VARLINK_DEFINE_ERROR(OnlyByDependency);
+
+static SD_VARLINK_DEFINE_ERROR(BusShuttingDown);
+
 SD_VARLINK_DEFINE_INTERFACE(
                 io_systemd_Manager,
                 "io.systemd.Manager",
@@ -191,7 +200,13 @@ SD_VARLINK_DEFINE_INTERFACE(
                 &vl_method_Reexecute,
                 SD_VARLINK_SYMBOL_COMMENT("Reload the manager configuration"),
                 &vl_method_Reload,
+                SD_VARLINK_SYMBOL_COMMENT("Enqueue all marked jobs"),
+                &vl_method_EnqueueMarkedJobs,
                 &vl_error_RateLimitReached,
+                SD_VARLINK_SYMBOL_COMMENT("Unit operation may be requested by dependency only"),
+                &vl_error_OnlyByDependency,
+                SD_VARLINK_SYMBOL_COMMENT("Operation refused, the bus is shutting down"),
+                &vl_error_BusShuttingDown,
                 &vl_type_ManagerContext,
                 &vl_type_ManagerRuntime,
                 &vl_type_Timestamp,

test/units/TEST-26-SYSTEMCTL.sh

@@ -376,6 +376,14 @@ systemctl show -P Markers "$UNIT_NAME" | grep needs-restart
 systemctl reload-or-restart --marked
 (! systemctl show -P Markers "$UNIT_NAME" | grep needs-restart)
 
+# again, but with varlinkctl instead
+systemctl restart "$UNIT_NAME"
+systemctl set-property "$UNIT_NAME" Markers=needs-restart
+systemctl show -P Markers "$UNIT_NAME" | grep needs-restart
+varlinkctl call /run/systemd/io.systemd.Manager io.systemd.Manager.EnqueueMarkedJobs '{}'
+timeout 30 bash -c "until systemctl list-jobs $UNIT_NAME | grep \"No jobs\" 2>/dev/null; do sleep 1; done"
+(! systemctl show -P Markers "$UNIT_NAME" | grep needs-restart)
+
 # --dry-run with destructive verbs
 # kexec is skipped intentionally, as it requires a bit more involved setup
 VERBS=(