vmspawn: dispatch QMP events in the bridge

The QMP client's userdata (the `VmspawnQmpBridge`, vmspawn's QPM state)
and its event callback were previously only set up in
`vmspawn_varlink_setup()`, after the VM has been resumed. This caused two
problems:

 - Boot-time device setup could not wait for its own completion before
   the VM is resumed, as that needs exactly these replies and events.

 - QMP replies and events that arrive before that were mishandled.
   Command callbacks like `on_qmp_complete()` crashed on a NULL bridge
   assertion. Also, `JOB_STATUS_CHANGE` events were dropped: they drive
   the boot-time blockdev-create continuation for `--ephemeral`.

   This only showed when the varlink setup failed, e.g. with a too long
   runtime directory path. Tearing down the bridge then failed the
   pending commands, and vmspawn aborted with an assertion instead of
   exiting with the error.

Both were introduced with commits 882049362d and 1a7b6fd9dd.

To fix this, set the `userdata` and bind the event callback when
creating the bridge. The bridge itself handles the events that drive its
own state (job continuations, `DEVICE_DELETED`), and forwards the others
to a callback that the varlink server binds for its event subscribers.

Signed-off-by: Martin Pitt <martin@amutable.com>
(cherry picked from commit 2cac293064215d3e4f5ed92a1e898f89e6ac4629)

src/shared/forward.h

@@ -97,3 +97,9 @@ typedef struct UnitInfo UnitInfo;
 typedef struct UserRecord UserRecord;
 typedef struct Verb Verb;
 typedef struct VeritySettings VeritySettings;
+
+typedef int (*qmp_event_callback_t)(
+                QmpClient *client,
+                const char *event,
+                sd_json_variant *data,
+                void *userdata);

src/shared/qmp-client.h

@@ -3,12 +3,6 @@
 
 #include "forward.h"
 
-typedef int (*qmp_event_callback_t)(
-                QmpClient *client,
-                const char *event,
-                sd_json_variant *data,
-                void *userdata);
-
 typedef void (*qmp_disconnect_callback_t)(
                 QmpClient *client,
                 void *userdata);

src/vmspawn/vmspawn-qmp.c

@@ -1099,7 +1099,7 @@ int vmspawn_qmp_remove_block_device(VmspawnQmpBridge *bridge, sd_varlink *link,
 
 /* DEVICE_DELETED arrives once the guest has acked the eject; only then is it
  * safe to drop the blockdev node and release the registry slot (and PCIe port). */
-int vmspawn_qmp_dispatch_device_deleted(VmspawnQmpBridge *bridge, sd_json_variant *data) {
+static int vmspawn_qmp_dispatch_device_deleted(VmspawnQmpBridge *bridge, sd_json_variant *data) {
         assert(bridge);
 
         if (!data)
@@ -1864,6 +1864,96 @@ static int probe_schema(QmpClient *c, VmspawnQmpBridge *bridge) {
                         on_probe_schema_reply, bridge);
 }
 
+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, /* ret_slot= */ NULL, "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) {
+
+        VmspawnQmpBridge *bridge = ASSERT_PTR(userdata);
+
+        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(bridge, data);
+
+        /* Notification still fans out below. */
+        if (streq(event, "DEVICE_DELETED"))
+                (void) vmspawn_qmp_dispatch_device_deleted(bridge, data);
+
+        if (!bridge->event_callback)
+                return 0;
+
+        return bridge->event_callback(client, event, data, bridge->event_userdata);
+}
+
+void vmspawn_qmp_bridge_bind_event(VmspawnQmpBridge *b, qmp_event_callback_t callback, void *userdata) {
+        assert(b);
+
+        b->event_callback = callback;
+        b->event_userdata = userdata;
+}
+
 int vmspawn_qmp_init(VmspawnQmpBridge **ret, int fd, sd_event *event) {
         _cleanup_(vmspawn_qmp_bridge_freep) VmspawnQmpBridge *bridge = NULL;
         _cleanup_close_ int fd_close = ASSERT_FD(TAKE_FD(fd));
@@ -1888,6 +1978,10 @@ int vmspawn_qmp_init(VmspawnQmpBridge **ret, int fd, sd_event *event) {
         if (r < 0)
                 return log_error_errno(r, "Failed to set QMP client description: %m");
 
+        /* Command callbacks find the bridge through the client's userdata. */
+        qmp_client_set_userdata(bridge->qmp, bridge);
+        qmp_client_bind_event(bridge->qmp, on_qmp_event, bridge);
+
         r = qmp_client_attach_event(bridge->qmp, event, SD_EVENT_PRIORITY_NORMAL);
         if (r < 0)
                 return log_error_errno(r, "Failed to attach QMP client to event loop: %m");

src/vmspawn/vmspawn-qmp.h

@@ -35,6 +35,8 @@ typedef struct VmspawnQmpBridge {
         char *hotplug_port_owner[VMSPAWN_PCIE_HOTPLUG_SPARES];  /* owner id per port; NULL = free */
         int scsi_controller_port_idx;  /* hotplug port idx taken by virtio-scsi-pci, -1 if none */
         uint64_t next_block_counter;   /* monotonic counter feeding internal QMP names (vmspawn-<N>-*) */
+        qmp_event_callback_t event_callback;  /* receives the events the bridge doesn't consume itself */
+        void *event_userdata;
         VmspawnQmpFeatureFlags features;
         bool setup_done;
         bool scsi_controller_created;  /* virtio-scsi-pci has been device_add'd */
@@ -45,6 +47,8 @@ DEFINE_TRIVIAL_CLEANUP_FUNC(VmspawnQmpBridge *, vmspawn_qmp_bridge_free);
 
 QmpClient* vmspawn_qmp_bridge_get_qmp(VmspawnQmpBridge *b);
 
+void vmspawn_qmp_bridge_bind_event(VmspawnQmpBridge *b, qmp_event_callback_t callback, void *userdata);
+
 /* 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);
 
@@ -194,4 +198,3 @@ int vmspawn_qmp_replace_block_device(
                 const char *id,
                 int fd,
                 QmpDriveFlags fd_flags);
-int vmspawn_qmp_dispatch_device_deleted(VmspawnQmpBridge *bridge, sd_json_variant *data);

src/vmspawn/vmspawn-varlink.c

@@ -361,64 +361,6 @@ static void vl_disconnect(sd_varlink_server *server, sd_varlink *link, void *use
         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, /* ret_slot= */ NULL, "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 notify_event_subscribers(VmspawnVarlinkContext *ctx, const char *event_name, sd_json_variant *data) {
         _cleanup_(sd_json_variant_unrefp) sd_json_variant *notification = NULL;
         sd_varlink *link;
@@ -463,14 +405,6 @@ static int on_qmp_event(
         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);
-
-        /* Notification still fans out below. */
-        if (streq(event, "DEVICE_DELETED"))
-                (void) vmspawn_qmp_dispatch_device_deleted(ctx->bridge, data);
-
         return notify_event_subscribers(ctx, event, data);
 }
 
@@ -560,9 +494,8 @@ int vmspawn_varlink_setup(
                 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);
+        vmspawn_qmp_bridge_bind_event(ctx->bridge, on_qmp_event, ctx);
         qmp_client_bind_disconnect(ctx->bridge->qmp, on_qmp_disconnect, ctx);
-        qmp_client_set_userdata(ctx->bridge->qmp, ctx->bridge);
 
         log_debug("Varlink control server listening on %s", listen_address);