vmspawn: set up varlink bridge infrastructure

Create the QMP-to-varlink bridge layer (vmspawn-qmp.{c,h}) and the
varlink server layer (vmspawn-varlink.{c,h}).

The QMP bridge (VmspawnQmpBridge) owns the QmpClient connection and
manages pending background jobs (e.g. blockdev-create continuations).
vmspawn_qmp_init() creates the client and attaches it to the event
loop. vmspawn_qmp_probe_features() drives io_uring and qcow2
discard-no-unref probes synchronously via a qmp_client_process() +
qmp_client_wait() loop — the QMP handshake completes transparently on
the first invoke. vmspawn_qmp_start() resumes vCPUs via an async
"cont" command.

The varlink server (VmspawnVarlinkContext) exposes three interfaces:

- io.systemd.MachineInstance: generic machine control (Terminate,
  PowerOff, Reboot, Pause, Resume, Describe, SubscribeEvents).
  Method handlers forward to QMP commands asynchronously — the
  varlink reply is deferred until the QMP response arrives.
- io.systemd.VirtualMachineInstance: VM-specific (placeholder for
  future snapshot/migration methods).
- io.systemd.QemuMachineInstance: QEMU-specific (AcquireQMP stub).

The server listens on <runtime_dir>/control with mode 0600.

Event streaming follows the importd Pull pattern: SubscribeEvents
sends an initial {ready:true} notification, then fans out QMP events
to all subscribers. The disconnect handler only unrefs subscriber
links (matching resolved's vl_on_notification_disconnect pattern).

Introduce the MachineConfig aggregate in vmspawn-qmp.h grouping the
per-device info structures (DriveInfos, NetworkInfo, VirtiofsInfos,
VsockInfo) together with machine_config_done() that chains the
individual done helpers. Callers populate it field-by-field and rely
on the _cleanup_ attribute for orderly teardown regardless of which
device types the invocation ends up using.

Signed-off-by: Christian Brauner (Amutable) <brauner@kernel.org>

src/vmspawn/meson.build

@@ -7,6 +7,8 @@ endif
 vmspawn_sources = files(
         'vmspawn.c',
         'vmspawn-qemu-config.c',
+        'vmspawn-qmp.c',
+        'vmspawn-varlink.c',
         'vmspawn-settings.c',
         'vmspawn-scope.c',
         'vmspawn-mount.c',

src/vmspawn/vmspawn-qmp.h

@@ -0,0 +1,145 @@
+/* SPDX-License-Identifier: LGPL-2.1-or-later */
+#pragma once
+
+#include <net/ethernet.h>
+
+#include "shared-forward.h"
+
+/* Pending job continuation — called when a QMP background job reaches "concluded" state.
+ * Used by blockdev-create to chain remaining drive setup after the job completes. */
+typedef int (*pending_job_callback_t)(QmpClient *qmp, void *userdata);
+typedef void (*pending_job_free_t)(void *userdata);
+
+typedef struct PendingJob {
+        pending_job_callback_t on_concluded;
+        pending_job_free_t free_userdata;
+        void *userdata;
+} PendingJob;
+
+PendingJob* pending_job_free(PendingJob *j);
+DEFINE_TRIVIAL_CLEANUP_FUNC(PendingJob *, pending_job_free);
+
+typedef enum VmspawnQmpFeatureFlags {
+        VMSPAWN_QMP_FEATURE_IO_URING         = 1u << 0,
+        VMSPAWN_QMP_FEATURE_DISCARD_NO_UNREF = 1u << 1,
+} VmspawnQmpFeatureFlags;
+
+typedef struct VmspawnQmpBridge {
+        QmpClient *qmp;
+        Hashmap *pending_jobs;  /* job_id (string, owned) -> PendingJob* */
+        VmspawnQmpFeatureFlags features;
+} VmspawnQmpBridge;
+
+VmspawnQmpBridge* vmspawn_qmp_bridge_free(VmspawnQmpBridge *b);
+DEFINE_TRIVIAL_CLEANUP_FUNC(VmspawnQmpBridge *, vmspawn_qmp_bridge_free);
+
+QmpClient* vmspawn_qmp_bridge_get_qmp(VmspawnQmpBridge *b);
+
+/* Phase 1: Connect to VMM backend. Returns an opaque bridge ready for device setup. */
+int vmspawn_qmp_init(VmspawnQmpBridge **ret, int fd, sd_event *event);
+
+/* Phase 1b: Feature probing. Fires one-shot QMP commands and drives the client
+ * synchronously until every reply has been delivered. Populates bridge->features.
+ * Must run before the device-setup phase; both io_uring and discard-no-unref flags
+ * are consumed by vmspawn_qmp_setup_drives(). */
+int vmspawn_qmp_probe_features(VmspawnQmpBridge *bridge);
+
+/* Phase 3: Resume vCPUs. All commands are async — responses arrive during sd_event_loop(). */
+int vmspawn_qmp_start(VmspawnQmpBridge *bridge);
+
+int vmspawn_qmp_bridge_register_job(
+                VmspawnQmpBridge *b,
+                const char *job_id,
+                pending_job_callback_t on_concluded,
+                void *userdata,
+                pending_job_free_t free_userdata);
+
+typedef enum QmpDriveFlags {
+        QMP_DRIVE_BLOCK_DEVICE     = 1u << 0,
+        QMP_DRIVE_READ_ONLY        = 1u << 1,
+        QMP_DRIVE_DISCARD          = 1u << 2,
+        QMP_DRIVE_NO_FLUSH         = 1u << 3,
+        QMP_DRIVE_BOOT             = 1u << 4,
+        QMP_DRIVE_IO_URING         = 1u << 5,
+        QMP_DRIVE_DISCARD_NO_UNREF = 1u << 6,  /* qcow2 only */
+} QmpDriveFlags;
+
+/* Drive info for QMP-based drive setup */
+typedef struct DriveInfo {
+        const char *path;          /* kept for logging only — not passed to QEMU */
+        const char *format;        /* "raw" or "qcow2" */
+        const char *disk_driver;   /* "virtio-blk-pci", "scsi-hd", "scsi-cd", "nvme" */
+        char *serial;              /* owned */
+        char *node_name;           /* owned */
+        char *pcie_port;           /* owned: pcie-root-port id for device_add bus (NULL on non-PCIe) */
+        int fd;                    /* pre-opened image fd (owned, -EBADF if unused) */
+        int overlay_fd;            /* pre-opened anonymous overlay fd for ephemeral (owned, -EBADF if unused) */
+        QmpDriveFlags flags;
+} DriveInfo;
+
+void drive_info_done(DriveInfo *info);
+
+typedef struct DriveInfos {
+        DriveInfo *drives;
+        size_t n_drives;
+        char *scsi_pcie_port;  /* owned: pcie-root-port id for SCSI controller (NULL if no SCSI or non-PCIe) */
+} DriveInfos;
+
+void drive_infos_done(DriveInfos *infos);
+
+/* Network info for QMP-based network setup. Covers privileged TAP (by name),
+ * nsresourced TAP (by FD via getfd), and user-mode networking. The no-network
+ * case (-nic none) stays on the QEMU command line. */
+typedef struct NetworkInfo {
+        const char *type;                  /* "tap" or "user" — points to a string literal */
+        char *ifname;                      /* owned: TAP interface name (tap by name only, NULL if unset) */
+        struct ether_addr mac;             /* VM-side MAC address (tap only, valid iff mac_set) */
+        bool mac_set;
+        char *pcie_port;                   /* owned: pcie-root-port id for device_add bus (NULL on non-PCIe) */
+        int fd;                            /* TAP fd to pass via getfd (tap by fd only, -EBADF if unused) */
+} NetworkInfo;
+
+void network_info_done(NetworkInfo *info);
+
+/* Virtiofs device info for QMP-based chardev + device setup */
+typedef struct VirtiofsInfo {
+        char *id;              /* owned: chardev and device id (e.g. "rootdir", "mnt0") */
+        char *socket_path;     /* owned: virtiofsd listen socket path */
+        char *tag;             /* owned: virtiofs mount tag visible to guest */
+        char *pcie_port;       /* owned: pcie-root-port id for device_add bus (NULL on non-PCIe) */
+} VirtiofsInfo;
+
+void virtiofs_info_done(VirtiofsInfo *info);
+
+typedef struct VirtiofsInfos {
+        VirtiofsInfo *entries;
+        size_t n_entries;
+} VirtiofsInfos;
+
+void virtiofs_infos_done(VirtiofsInfos *infos);
+
+/* VSOCK device info for QMP-based setup via getfd + device_add */
+typedef struct VsockInfo {
+        int fd;                 /* vhost-vsock fd to pass via getfd (-EBADF if unused) */
+        unsigned cid;           /* guest CID */
+        char *pcie_port;        /* owned: pcie-root-port id for device_add bus (NULL on non-PCIe) */
+} VsockInfo;
+
+void vsock_info_done(VsockInfo *info);
+
+/* Aggregate of the per-device info structures populated before the bridge-based
+ * device setup phase. Keeps lifetime and cleanup of all device state in one place. */
+typedef struct MachineConfig {
+        DriveInfos drives;
+        NetworkInfo network;
+        VirtiofsInfos virtiofs;
+        VsockInfo vsock;
+} MachineConfig;
+
+void machine_config_done(MachineConfig *c);
+
+/* Phase 2: Device setup — call any subset in any order before vmspawn_qmp_start(). */
+int vmspawn_qmp_setup_drives(VmspawnQmpBridge *bridge, DriveInfos *drives);
+int vmspawn_qmp_setup_network(VmspawnQmpBridge *bridge, NetworkInfo *network);
+int vmspawn_qmp_setup_virtiofs(VmspawnQmpBridge *bridge, const VirtiofsInfos *virtiofs);
+int vmspawn_qmp_setup_vsock(VmspawnQmpBridge *bridge, VsockInfo *vsock);

src/vmspawn/vmspawn-varlink.c

@@ -0,0 +1,410 @@
+/* SPDX-License-Identifier: LGPL-2.1-or-later */
+
+#include "alloc-util.h"
+#include "errno-util.h"
+#include "hashmap.h"
+#include "log.h"
+#include "path-util.h"
+#include "qmp-client.h"
+#include "string-util.h"
+#include "strv.h"
+#include "varlink-io.systemd.MachineInstance.h"
+#include "varlink-io.systemd.QemuMachineInstance.h"
+#include "varlink-io.systemd.VirtualMachineInstance.h"
+#include "varlink-util.h"
+#include "vmspawn-varlink.h"
+
+DEFINE_PRIVATE_HASH_OPS_FULL(
+                varlink_subscriber_hash_ops,
+                void, trivial_hash_func, trivial_compare_func, sd_varlink_close_unref,
+                char*, strv_free);
+
+struct VmspawnVarlinkContext {
+        sd_varlink_server *varlink_server;
+        VmspawnQmpBridge *bridge;
+        /* Key: sd_varlink* (ref'd), Value: strv filter (NULL = all events).
+         * varlink_subscriber_hash_ops handles cleanup of both on removal. */
+        Hashmap *subscribed;
+};
+
+/* Translate a QMP async completion into a varlink error reply */
+static int qmp_error_to_varlink(sd_varlink *link, const char *error_desc, int error) {
+        assert(link);
+
+        if (ERRNO_IS_DISCONNECT(error))
+                return sd_varlink_error(link, "io.systemd.MachineInstance.NotConnected", NULL);
+        if (error == -EIO)
+                log_warning("QMP command failed: %s", strna(error_desc));
+        return sd_varlink_error_errno(link, error);
+}
+
+/* Shared async completion for simple QMP commands that return no data.
+ * Errors are translated to varlink replies, not propagated through sd_event. */
+static int on_qmp_simple_complete(
+                QmpClient *client,
+                sd_json_variant *result,
+                const char *error_desc,
+                int error,
+                void *userdata) {
+
+        sd_varlink *link = ASSERT_PTR(userdata);
+
+        assert(client);
+
+        if (error == 0)
+                (void) sd_varlink_reply(link, NULL);
+        else
+                (void) qmp_error_to_varlink(link, error_desc, error);
+
+        sd_varlink_unref(link);
+        return 0;
+}
+
+static int qmp_execute_varlink_async(
+                VmspawnVarlinkContext *ctx,
+                sd_varlink *link,
+                const char *command,
+                sd_json_variant *arguments,
+                qmp_command_callback_t callback) {
+
+        int r;
+
+        sd_varlink_ref(link);
+
+        r = qmp_client_invoke(ctx->bridge->qmp, command, QMP_CLIENT_ARGS(arguments), callback, link);
+        if (r < 0)
+                sd_varlink_unref(link);
+
+        return r;
+}
+
+static int qmp_execute_simple_async(sd_varlink *link, VmspawnVarlinkContext *ctx, const char *qmp_command) {
+        assert(link);
+        assert(ctx);
+        assert(qmp_command);
+
+        return qmp_execute_varlink_async(ctx, link, qmp_command, /* arguments= */ NULL, on_qmp_simple_complete);
+}
+
+static int vl_method_terminate(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata) {
+        return qmp_execute_simple_async(link, ASSERT_PTR(userdata), "quit");
+}
+
+static int vl_method_pause(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata) {
+        return qmp_execute_simple_async(link, ASSERT_PTR(userdata), "stop");
+}
+
+static int vl_method_resume(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata) {
+        return qmp_execute_simple_async(link, ASSERT_PTR(userdata), "cont");
+}
+
+static int vl_method_power_off(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata) {
+        return qmp_execute_simple_async(link, ASSERT_PTR(userdata), "system_powerdown");
+}
+
+static int vl_method_reboot(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata) {
+        return qmp_execute_simple_async(link, ASSERT_PTR(userdata), "system_reset");
+}
+
+/* Async completion for query-status: extract running/status from QMP result */
+static int on_qmp_describe_complete(
+                QmpClient *client,
+                sd_json_variant *result,
+                const char *error_desc,
+                int error,
+                void *userdata) {
+
+        _cleanup_(sd_varlink_unrefp) sd_varlink *link = ASSERT_PTR(userdata);
+
+        assert(client);
+
+        if (error != 0) {
+                (void) qmp_error_to_varlink(link, error_desc, error);
+                return 0;
+        }
+
+        sd_json_variant *running = sd_json_variant_by_key(result, "running");
+        sd_json_variant *status = sd_json_variant_by_key(result, "status");
+
+        (void) sd_varlink_replybo(
+                        link,
+                        SD_JSON_BUILD_PAIR_BOOLEAN("running", running ? sd_json_variant_boolean(running) : false),
+                        SD_JSON_BUILD_PAIR_STRING("status", status && sd_json_variant_is_string(status) ? sd_json_variant_string(status) : "unknown"));
+
+        return 0;
+}
+
+static int vl_method_describe(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata) {
+        VmspawnVarlinkContext *ctx = ASSERT_PTR(userdata);
+
+        return qmp_execute_varlink_async(ctx, link, "query-status", /* arguments= */ NULL, on_qmp_describe_complete);
+}
+
+static int vl_method_subscribe_events(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata) {
+        VmspawnVarlinkContext *ctx = ASSERT_PTR(userdata);
+        _cleanup_strv_free_ char **filter = NULL;
+        int r;
+
+        /* SD_VARLINK_REQUIRES_MORE in the IDL rejects non-streaming callers before we get here */
+
+        r = sd_varlink_dispatch(link, parameters, (const sd_json_dispatch_field[]) {
+                { "filter", _SD_JSON_VARIANT_TYPE_INVALID, sd_json_dispatch_strv, 0, SD_JSON_NULLABLE },
+                {},
+        }, &filter);
+        if (r != 0)
+                return r;
+
+        sd_varlink_ref(link);
+
+        r = hashmap_ensure_put(&ctx->subscribed, &varlink_subscriber_hash_ops, link, filter);
+        if (r < 0) {
+                sd_varlink_unref(link);
+                return r;
+        }
+
+        TAKE_PTR(filter);
+
+        r = sd_varlink_notifybo(link, SD_JSON_BUILD_PAIR_STRING("event", "READY"));
+        if (r < 0) {
+                strv_free(hashmap_remove(ctx->subscribed, link));
+                sd_varlink_close_unref(link);
+                return r;
+        }
+
+        return 0;
+}
+
+static int vl_method_acquire_qmp(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata) {
+        return sd_varlink_error_errno(link, -EOPNOTSUPP);
+}
+
+static void vl_disconnect(sd_varlink_server *server, sd_varlink *link, void *userdata) {
+        VmspawnVarlinkContext *ctx = ASSERT_PTR(userdata);
+
+        assert(server);
+        assert(link);
+
+        /* Only subscribers hold an extra ref on the link (taken in vl_method_subscribe_events).
+         * Non-subscriber connections (one-shot commands like Pause, Describe) must not be unref'd
+         * here — their extra ref is consumed by the async completion callback. Only unref, never
+         * close — the server handles close after this callback returns (matching resolved's
+         * vl_on_notification_disconnect pattern).
+         *
+         * Use hashmap_remove2() so the returned key (non-NULL iff the entry was present)
+         * disambiguates "no filter subscriber" (value=NULL) from "not a subscriber". */
+        void *removed_key = NULL;
+        strv_free(hashmap_remove2(ctx->subscribed, link, &removed_key));
+        if (!removed_key)
+                return;
+
+        sd_varlink_unref(link);
+}
+
+static int on_job_dismiss_complete(
+                QmpClient *client,
+                sd_json_variant *result,
+                const char *error_desc,
+                int error,
+                void *userdata) {
+
+        if (error < 0)
+                log_debug_errno(error, "job-dismiss failed: %s", strna(error_desc));
+
+        return 0;
+}
+
+static int dispatch_pending_job(VmspawnQmpBridge *bridge, sd_json_variant *data) {
+        const char *job_id, *status;
+        int r;
+
+        assert(bridge);
+
+        if (!data)
+                return 0;
+
+        job_id = sd_json_variant_string(sd_json_variant_by_key(data, "id"));
+        status = sd_json_variant_string(sd_json_variant_by_key(data, "status"));
+
+        if (!job_id || !streq_ptr(status, "concluded"))
+                return 0;
+
+        _cleanup_free_ char *key = NULL;
+        _cleanup_(pending_job_freep) PendingJob *job = hashmap_remove2(bridge->pending_jobs, job_id, (void**) &key);
+        if (!job)
+                return 0;
+
+        log_debug("QMP job '%s' concluded, firing continuation", job_id);
+
+        /* Dismiss the concluded job before running the continuation */
+        _cleanup_(sd_json_variant_unrefp) sd_json_variant *dismiss_args = NULL;
+        r = sd_json_buildo(&dismiss_args, SD_JSON_BUILD_PAIR_STRING("id", job_id));
+        if (r < 0)
+                return sd_event_exit(qmp_client_get_event(bridge->qmp), r);
+
+        r = qmp_client_invoke(bridge->qmp, "job-dismiss", QMP_CLIENT_ARGS(dismiss_args),
+                              on_job_dismiss_complete, /* userdata= */ NULL);
+        if (r < 0)
+                return sd_event_exit(qmp_client_get_event(bridge->qmp), r);
+
+        if (!job->on_concluded)
+                return 1;
+
+        r = job->on_concluded(bridge->qmp, TAKE_PTR(job->userdata));
+        if (r < 0) {
+                log_error_errno(r, "Job continuation failed: %m");
+                return sd_event_exit(qmp_client_get_event(bridge->qmp), r);
+        }
+
+        return 1;
+}
+
+static int on_qmp_event(
+                QmpClient *client,
+                const char *event,
+                sd_json_variant *data,
+                void *userdata) {
+
+        VmspawnVarlinkContext *ctx = ASSERT_PTR(userdata);
+        _cleanup_(sd_json_variant_unrefp) sd_json_variant *notification = NULL;
+        sd_varlink *link;
+        char **filter;
+        int r;
+
+        assert(client);
+        assert(event);
+
+        /* Dispatch job status changes to pending continuations (e.g. blockdev-create) */
+        if (streq(event, "JOB_STATUS_CHANGE"))
+                return dispatch_pending_job(ctx->bridge, data);
+
+        if (hashmap_isempty(ctx->subscribed))
+                return 0;
+
+        r = sd_json_buildo(
+                        &notification,
+                        SD_JSON_BUILD_PAIR_STRING("event", event),
+                        SD_JSON_BUILD_PAIR_CONDITION(!!data, "data", SD_JSON_BUILD_VARIANT(data)));
+        if (r < 0) {
+                log_warning_errno(r, "Failed to build event notification, ignoring: %m");
+                return 0;
+        }
+
+        HASHMAP_FOREACH_KEY(filter, link, ctx->subscribed) {
+                if (filter && !strv_contains(filter, event))
+                        continue;
+
+                r = sd_varlink_notify(link, notification);
+                if (r < 0)
+                        log_warning_errno(r, "Failed to notify event subscriber, ignoring: %m");
+        }
+
+        return 0;
+}
+
+/* Free all subscriber entries — varlink_subscriber_hash_ops handles
+ * close + unref for each key and strv_free for each value. */
+static void drain_event_subscribers(Hashmap **subscribed) {
+        assert(subscribed);
+        *subscribed = hashmap_free(*subscribed);
+}
+
+static void on_qmp_disconnect(QmpClient *client, void *userdata) {
+        VmspawnVarlinkContext *ctx = ASSERT_PTR(userdata);
+
+        assert(client);
+
+        log_debug("Backend connection lost");
+
+        /* Propagate connection loss by closing all subscriber connections */
+        drain_event_subscribers(&ctx->subscribed);
+}
+
+int vmspawn_varlink_setup(
+                VmspawnVarlinkContext **ret,
+                VmspawnQmpBridge *bridge,
+                const char *runtime_dir,
+                char **ret_control_address) {
+
+        _cleanup_(vmspawn_varlink_context_freep) VmspawnVarlinkContext *ctx = NULL;
+        _cleanup_free_ char *listen_address = NULL;
+        int r;
+
+        assert(ret);
+        assert(bridge);
+        assert(runtime_dir);
+
+        sd_event *event = qmp_client_get_event(bridge->qmp);
+        assert(event);
+
+        ctx = new0(VmspawnVarlinkContext, 1);
+        if (!ctx)
+                return log_oom();
+
+        /* Create varlink server for VM control */
+        r = varlink_server_new(&ctx->varlink_server,
+                               SD_VARLINK_SERVER_INHERIT_USERDATA,
+                               ctx);
+        if (r < 0)
+                return log_error_errno(r, "Failed to create varlink server: %m");
+
+        r = sd_varlink_server_add_interface_many(
+                        ctx->varlink_server,
+                        &vl_interface_io_systemd_MachineInstance,
+                        &vl_interface_io_systemd_VirtualMachineInstance,
+                        &vl_interface_io_systemd_QemuMachineInstance);
+        if (r < 0)
+                return log_error_errno(r, "Failed to add varlink interfaces: %m");
+
+        r = sd_varlink_server_bind_method_many(
+                        ctx->varlink_server,
+                        "io.systemd.MachineInstance.Terminate",         vl_method_terminate,
+                        "io.systemd.MachineInstance.PowerOff",          vl_method_power_off,
+                        "io.systemd.MachineInstance.Pause",             vl_method_pause,
+                        "io.systemd.MachineInstance.Resume",            vl_method_resume,
+                        "io.systemd.MachineInstance.Reboot",            vl_method_reboot,
+                        "io.systemd.MachineInstance.Describe",          vl_method_describe,
+                        "io.systemd.MachineInstance.SubscribeEvents",   vl_method_subscribe_events,
+                        "io.systemd.QemuMachineInstance.AcquireQMP",    vl_method_acquire_qmp);
+        if (r < 0)
+                return log_error_errno(r, "Failed to bind varlink methods: %m");
+
+        r = sd_varlink_server_bind_disconnect(ctx->varlink_server, vl_disconnect);
+        if (r < 0)
+                return log_error_errno(r, "Failed to bind disconnect handler: %m");
+
+        listen_address = path_join(runtime_dir, "control");
+        if (!listen_address)
+                return log_oom();
+
+        r = sd_varlink_server_listen_address(ctx->varlink_server, listen_address, 0600);
+        if (r < 0)
+                return log_error_errno(r, "Failed to listen on %s: %m", listen_address);
+
+        r = sd_varlink_server_attach_event(ctx->varlink_server, event, SD_EVENT_PRIORITY_NORMAL);
+        if (r < 0)
+                return log_error_errno(r, "Failed to attach varlink server to event loop: %m");
+
+        ctx->bridge = bridge;
+        qmp_client_bind_event(ctx->bridge->qmp, on_qmp_event, ctx);
+        qmp_client_bind_disconnect(ctx->bridge->qmp, on_qmp_disconnect, ctx);
+
+        log_debug("Varlink control server listening on %s", listen_address);
+
+        if (ret_control_address)
+                *ret_control_address = TAKE_PTR(listen_address);
+
+        *ret = TAKE_PTR(ctx);
+        return 0;
+}
+
+VmspawnVarlinkContext* vmspawn_varlink_context_free(VmspawnVarlinkContext *ctx) {
+        if (!ctx)
+                return NULL;
+
+        sd_varlink_server_unref(ctx->varlink_server);
+        vmspawn_qmp_bridge_free(ctx->bridge);
+
+        drain_event_subscribers(&ctx->subscribed);
+
+        return mfree(ctx);
+}

src/vmspawn/vmspawn-varlink.h

@@ -0,0 +1,19 @@
+/* SPDX-License-Identifier: LGPL-2.1-or-later */
+#pragma once
+
+#include "cleanup-util.h"
+#include "shared-forward.h"
+#include "vmspawn-qmp.h"
+
+typedef struct VmspawnVarlinkContext VmspawnVarlinkContext;
+
+/* Varlink server for VM control on top of an established bridge connection */
+int vmspawn_varlink_setup(
+                VmspawnVarlinkContext **ret,
+                VmspawnQmpBridge *bridge,
+                const char *runtime_dir,
+                char **ret_control_address);
+
+VmspawnVarlinkContext* vmspawn_varlink_context_free(VmspawnVarlinkContext *ctx);
+
+DEFINE_TRIVIAL_CLEANUP_FUNC(VmspawnVarlinkContext *, vmspawn_varlink_context_free);