qmp-client: add synchronous qmp_client_call()

Add a synchronous counterpart to qmp_client_invoke() that pumps the
client's own process()/wait() loop until the reply for the issued
command id arrives, mirroring sd_varlink_call()'s contract: *ret_result
and *ret_error_desc are borrowed pointers into c->current, valid until
the next qmp_client_call(), and a QMP error surfaces as -EIO when the
caller doesn't ask for the description.

Factor the command-build + slot-insert + enqueue sequence shared with
qmp_client_invoke() into qmp_client_send(). A NULL callback marks the
slot as synchronous: dispatch_reply still matches on id (so unknown
ids continue to be logged and discarded, preserving async-only
robustness), but skips the TAKE_PTR + callback invocation and leaves
c->current pinned for qmp_client_call() to read out.

Cover the three paths in test-qmp-client: successful reply, QMP error
with ret_error_desc, and QMP error returned as -EIO.

src/shared/qmp-client.c

@@ -232,6 +232,11 @@ static int qmp_client_dispatch_reply(QmpClient *c) {
                 return 1;
         }
 
+        /* Synchronous slot (no callback): leave c->current pinned so qmp_client_call() can
+         * pick up the reply and hand out borrowed pointers into it. */
+        if (!pending->callback)
+                return 1;
+
         _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = TAKE_PTR(c->current);
         error = qmp_parse_response(v, &result, &desc);
 
@@ -250,9 +255,11 @@ static void qmp_client_fail_pending(QmpClient *c, int error) {
         assert(c);
 
         while ((p = set_steal_first(c->slots))) {
-                r = p->callback(c, /* result= */ NULL, /* error_desc= */ NULL, error, p->userdata);
-                if (r < 0)
-                        json_stream_log_errno(&c->stream, r, "Command callback returned error, ignoring: %m");
+                if (p->callback) {
+                        r = p->callback(c, /* result= */ NULL, /* error_desc= */ NULL, error, p->userdata);
+                        if (r < 0)
+                                json_stream_log_errno(&c->stream, r, "Command callback returned error, ignoring: %m");
+                }
                 free(p);
         }
 }
@@ -692,12 +699,16 @@ static QmpClientArgs* qmp_client_args_close_fds(QmpClientArgs *p) {
 
 DEFINE_TRIVIAL_CLEANUP_FUNC(QmpClientArgs*, qmp_client_args_close_fds);
 
-int qmp_client_invoke(
+/* Shared send path for qmp_client_invoke() and qmp_client_call(). A NULL callback registers
+ * a "synchronous" slot: dispatch_reply leaves c->current pinned on match instead of invoking
+ * a callback, so qmp_client_call() can hand out borrowed pointers into the reply. */
+static int qmp_client_send(
                 QmpClient *c,
                 const char *command,
                 QmpClientArgs *args,
                 qmp_command_callback_t callback,
-                void *userdata) {
+                void *userdata,
+                uint64_t *ret_id) {
 
         _cleanup_(sd_json_variant_unrefp) sd_json_variant *cmd = NULL;
         _cleanup_free_ QmpSlot *pending = NULL;
@@ -709,7 +720,6 @@ int qmp_client_invoke(
 
         assert(c);
         assert(command);
-        assert(callback);
 
         r = qmp_client_ensure_running(c);
         if (r < 0)
@@ -748,9 +758,83 @@ int qmp_client_invoke(
 
         TAKE_PTR(pending);
         TAKE_PTR(fds_owner);
+
+        if (ret_id)
+                *ret_id = id;
         return 0;
 }
 
+int qmp_client_invoke(
+                QmpClient *c,
+                const char *command,
+                QmpClientArgs *args,
+                qmp_command_callback_t callback,
+                void *userdata) {
+
+        assert(callback);
+        return qmp_client_send(c, command, args, callback, userdata, /* ret_id= */ NULL);
+}
+
+int qmp_client_call(
+                QmpClient *c,
+                const char *command,
+                QmpClientArgs *args,
+                sd_json_variant **ret_result,
+                const char **ret_error_desc) {
+
+        uint64_t id;
+        int r;
+
+        assert_return(c, -EINVAL);
+        assert_return(command, -EINVAL);
+
+        /* Drop any reply pinned by a previous qmp_client_call() before we pin a new one. */
+        qmp_client_clear_current(c);
+
+        /* NULL callback marks this as a synchronous slot: dispatch_reply matches on id like
+         * any other slot (so stray unknown-id replies still get logged and dropped), but
+         * pins c->current for us instead of invoking a callback. */
+        r = qmp_client_send(c, command, args, /* callback= */ NULL, /* userdata= */ NULL, &id);
+        if (r < 0)
+                return r;
+
+        /* Pump the loop until our sync slot fires (removed from c->slots, c->current pinned). */
+        for (;;) {
+                if (c->state == QMP_CLIENT_DISCONNECTED)
+                        return -ECONNRESET;
+
+                if (!set_contains(c->slots, &(QmpSlot) { .id = id })) {
+                        assert(c->current);
+                        break;
+                }
+
+                r = qmp_client_process(c);
+                if (r < 0)
+                        return r;
+                if (r > 0)
+                        continue;
+
+                r = qmp_client_wait(c, USEC_INFINITY);
+                if (r < 0)
+                        return r;
+        }
+
+        sd_json_variant *result = NULL;
+        const char *desc = NULL;
+        int error = qmp_parse_response(c->current, &result, &desc);
+
+        /* If caller doesn't ask for the error string, surface the error as the return code. */
+        if (!ret_error_desc && error < 0)
+                return error;
+
+        if (ret_result)
+                *ret_result = result;
+        if (ret_error_desc)
+                *ret_error_desc = desc;
+
+        return 1;
+}
+
 void qmp_client_bind_event(QmpClient *c, qmp_event_callback_t callback, void *userdata) {
         assert(c);
         c->event_callback = callback;

src/shared/qmp-client.h

@@ -62,6 +62,16 @@ int qmp_client_invoke(
                 qmp_command_callback_t callback,
                 void *userdata);
 
+/* Synchronous send + receive. Pumps the event loop until the reply arrives. *ret_result and
+ * *ret_error_desc are borrowed pointers into the last reply, valid until the next
+ * qmp_client_call(). Same contract as sd_varlink_call(). */
+int qmp_client_call(
+                QmpClient *client,
+                const char *command,
+                QmpClientArgs *args,
+                sd_json_variant **ret_result,
+                const char **ret_error_desc);
+
 void qmp_client_bind_event(QmpClient *c, qmp_event_callback_t callback, void *userdata);
 void qmp_client_bind_disconnect(QmpClient *c, qmp_disconnect_callback_t callback, void *userdata);
 int qmp_client_set_description(QmpClient *c, const char *description);

src/test/test-qmp-client.c

@@ -473,6 +473,148 @@ TEST(qmp_client_invoke_failure_closes_fds) {
         ASSERT_EQ(errno, EBADF);
 }
 
+/* Reads one command, asserts its execute name, and replies with a QMP error object carrying
+ * the given description. Mirrors mock_qmp_expect_and_reply() but on the error branch. */
+static void mock_qmp_expect_and_reply_error(int fd, const char *expected_command, const char *error_desc) {
+        _cleanup_free_ char *buf = NULL;
+        _cleanup_(sd_json_variant_unrefp) sd_json_variant *cmd = NULL, *error_obj = NULL, *response = NULL;
+
+        buf = ASSERT_NOT_NULL(new(char, 4096));
+
+        ssize_t n = read(fd, buf, 4095);
+        assert_se(n > 0);
+        buf[n] = '\0';
+
+        ASSERT_OK(sd_json_parse(buf, 0, &cmd, NULL, NULL));
+
+        sd_json_variant *execute = ASSERT_NOT_NULL(sd_json_variant_by_key(cmd, "execute"));
+        ASSERT_STREQ(sd_json_variant_string(execute), expected_command);
+
+        sd_json_variant *id = ASSERT_NOT_NULL(sd_json_variant_by_key(cmd, "id"));
+
+        ASSERT_OK(sd_json_buildo(
+                        &error_obj,
+                        SD_JSON_BUILD_PAIR_STRING("class", "GenericError"),
+                        SD_JSON_BUILD_PAIR_STRING("desc", error_desc)));
+
+        ASSERT_OK(sd_json_buildo(
+                        &response,
+                        SD_JSON_BUILD_PAIR("error", SD_JSON_BUILD_VARIANT(error_obj)),
+                        SD_JSON_BUILD_PAIR("id", SD_JSON_BUILD_VARIANT(id))));
+
+        mock_qmp_write_json(fd, response);
+}
+
+/* Drives a small wire dance for the sync call test: greeting, capabilities, one successful
+ * command reply, and two error replies (one for the ret_error_desc path, one for the -EIO
+ * path). */
+static _noreturn_ void mock_qmp_server_call(int fd) {
+        _cleanup_(sd_json_variant_unrefp) sd_json_variant *status_return = NULL;
+
+        mock_qmp_write_literal(fd,
+                "{\"QMP\": {\"version\": {\"qemu\": {\"micro\": 0, \"minor\": 0, \"major\": 9}}, \"capabilities\": []}}");
+
+        mock_qmp_expect_and_reply(fd, "qmp_capabilities", NULL);
+
+        ASSERT_OK(sd_json_buildo(
+                        &status_return,
+                        SD_JSON_BUILD_PAIR_BOOLEAN("running", true),
+                        SD_JSON_BUILD_PAIR_STRING("status", "running")));
+        mock_qmp_expect_and_reply(fd, "query-status", status_return);
+
+        mock_qmp_expect_and_reply_error(fd, "stop", "not running");
+        mock_qmp_expect_and_reply_error(fd, "stop", "still not running");
+
+        safe_close(fd);
+        _exit(EXIT_SUCCESS);
+}
+
+TEST(qmp_client_call) {
+        _cleanup_(qmp_client_unrefp) QmpClient *client = NULL;
+        _cleanup_(pidref_done_sigkill_wait) PidRef pid = PIDREF_NULL;
+        int qmp_fds[2];
+        int r;
+
+        ASSERT_OK_ERRNO(socketpair(AF_UNIX, SOCK_STREAM|SOCK_CLOEXEC, 0, qmp_fds));
+
+        r = ASSERT_OK(pidref_safe_fork("(mock-qmp-call)", FORK_DEATHSIG_SIGKILL|FORK_LOG, &pid));
+        if (r == 0) {
+                safe_close(qmp_fds[0]);
+                mock_qmp_server_call(qmp_fds[1]);
+        }
+        safe_close(qmp_fds[1]);
+
+        /* qmp_client_call() drives its own process()+wait() pump, so no event loop needed. */
+        ASSERT_OK(qmp_client_connect_fd(&client, qmp_fds[0]));
+
+        /* Successful call: borrowed result pointer is valid until the next call. */
+        sd_json_variant *result = NULL;
+        const char *error_desc = NULL;
+        ASSERT_EQ(qmp_client_call(client, "query-status", NULL, &result, &error_desc), 1);
+        ASSERT_NULL(error_desc);
+        ASSERT_NOT_NULL(result);
+
+        sd_json_variant *running = ASSERT_NOT_NULL(sd_json_variant_by_key(result, "running"));
+        ASSERT_TRUE(sd_json_variant_boolean(running));
+        sd_json_variant *status = ASSERT_NOT_NULL(sd_json_variant_by_key(result, "status"));
+        ASSERT_STREQ(sd_json_variant_string(status), "running");
+
+        /* QMP error with ret_error_desc provided: returns 1, result NULL, desc set. */
+        result = (sd_json_variant*) 0x1;  /* poison to catch lack-of-write */
+        error_desc = NULL;
+        ASSERT_EQ(qmp_client_call(client, "stop", NULL, &result, &error_desc), 1);
+        ASSERT_NULL(result);
+        ASSERT_STREQ(error_desc, "not running");
+
+        /* QMP error without ret_error_desc: surfaces as -EIO. */
+        ASSERT_EQ(qmp_client_call(client, "stop", NULL, NULL, NULL), -EIO);
+}
+
+/* Server variant for the sync-call disconnect test: greets, accepts capabilities, reads one
+ * command without replying, then closes the socket so the client sees EOF mid-wait. */
+static _noreturn_ void mock_qmp_server_call_disconnect(int fd) {
+        _cleanup_free_ char *buf = NULL;
+
+        mock_qmp_write_literal(fd,
+                "{\"QMP\": {\"version\": {\"qemu\": {\"micro\": 0, \"minor\": 0, \"major\": 9}}, \"capabilities\": []}}");
+
+        mock_qmp_expect_and_reply(fd, "qmp_capabilities", NULL);
+
+        /* Consume the stop command but don't reply — just close to trigger EOF while the
+         * client is blocked in qmp_client_call()'s process+wait pump. */
+        buf = ASSERT_NOT_NULL(new(char, 4096));
+        ssize_t n = read(fd, buf, 4095);
+        assert_se(n > 0);
+
+        safe_close(fd);
+        _exit(EXIT_SUCCESS);
+}
+
+TEST(qmp_client_call_disconnect) {
+        _cleanup_(qmp_client_unrefp) QmpClient *client = NULL;
+        _cleanup_(pidref_done_sigkill_wait) PidRef pid = PIDREF_NULL;
+        int qmp_fds[2];
+        int r;
+
+        ASSERT_OK_ERRNO(socketpair(AF_UNIX, SOCK_STREAM|SOCK_CLOEXEC, 0, qmp_fds));
+
+        r = ASSERT_OK(pidref_safe_fork("(mock-qmp-call-disc)", FORK_DEATHSIG_SIGKILL|FORK_LOG, &pid));
+        if (r == 0) {
+                safe_close(qmp_fds[0]);
+                mock_qmp_server_call_disconnect(qmp_fds[1]);
+        }
+        safe_close(qmp_fds[1]);
+
+        ASSERT_OK(qmp_client_connect_fd(&client, qmp_fds[0]));
+
+        /* The server reads our stop command and closes without replying. qmp_client_call()
+         * is driving its own pump, so it must notice the EOF, transition to DISCONNECTED,
+         * and return a disconnect error rather than hanging. */
+        r = qmp_client_call(client, "stop", NULL, NULL, NULL);
+        ASSERT_TRUE(r < 0);
+        ASSERT_TRUE(ERRNO_IS_NEG_DISCONNECT(r));
+}
+
 TEST(qmp_schema_has_member) {
         _cleanup_(sd_json_variant_unrefp) sd_json_variant *schema = NULL;