sd-varlink: add sd_varlink_invoke_and_upgrade()

Add the asynchronous counterpart of sd_varlink_call_and_upgrade(): the
method call is enqueued with the upgrade request, the reply is delivered
to the reply callback as usual, and if the server accepted the upgrade
the connection's file descriptors are handed to the callback installed
with sd_varlink_bind_upgrade(), mirroring how the server side works. If
the server replied with an error the connection stays usable.

A new client state VARLINK_AWAITING_REPLY_UPGRADE tracks the pending
request; no other calls may be pipelined behind it.

man/rules/meson.build

@@ -974,7 +974,11 @@ manpages = [
   'HAVE_PAM'],
  ['sd_varlink_call_and_upgrade',
   '3',
-  ['sd_varlink_call_and_upgradeb', 'sd_varlink_call_and_upgradebo'],
+  ['sd_varlink_call_and_upgradeb',
+   'sd_varlink_call_and_upgradebo',
+   'sd_varlink_invoke_and_upgrade',
+   'sd_varlink_invoke_and_upgradeb',
+   'sd_varlink_invoke_and_upgradebo'],
   ''],
  ['sd_varlink_connect_address',
   '3',

man/sd_varlink_call_and_upgrade.xml

@@ -19,6 +19,9 @@
     <refname>sd_varlink_call_and_upgrade</refname>
     <refname>sd_varlink_call_and_upgradeb</refname>
     <refname>sd_varlink_call_and_upgradebo</refname>
+    <refname>sd_varlink_invoke_and_upgrade</refname>
+    <refname>sd_varlink_invoke_and_upgradeb</refname>
+    <refname>sd_varlink_invoke_and_upgradebo</refname>
 
     <refpurpose>Invoke a Varlink method and take over the connection for a raw protocol</refpurpose>
   </refnamediv>
@@ -59,6 +62,27 @@
         <paramdef>int *<parameter>ret_output_fd</parameter></paramdef>
         <paramdef>…</paramdef>
       </funcprototype>
+
+      <funcprototype>
+        <funcdef>int <function>sd_varlink_invoke_and_upgrade</function></funcdef>
+        <paramdef>sd_varlink *<parameter>link</parameter></paramdef>
+        <paramdef>const char *<parameter>method</parameter></paramdef>
+        <paramdef>sd_json_variant *<parameter>parameters</parameter></paramdef>
+      </funcprototype>
+
+      <funcprototype>
+        <funcdef>int <function>sd_varlink_invoke_and_upgradeb</function></funcdef>
+        <paramdef>sd_varlink *<parameter>link</parameter></paramdef>
+        <paramdef>const char *<parameter>method</parameter></paramdef>
+        <paramdef>…</paramdef>
+      </funcprototype>
+
+      <funcprototype>
+        <funcdef>int <function>sd_varlink_invoke_and_upgradebo</function></funcdef>
+        <paramdef>sd_varlink *<parameter>link</parameter></paramdef>
+        <paramdef>const char *<parameter>method</parameter></paramdef>
+        <paramdef>…</paramdef>
+      </funcprototype>
     </funcsynopsis>
   </refsynopsisdiv>
 
@@ -111,14 +135,36 @@
     relationship mirrors that of
     <citerefentry><refentrytitle>sd_varlink_call</refentrytitle><manvolnum>3</manvolnum></citerefentry> and its
     <function>sd_varlink_callb()</function>/<function>sd_varlink_callbo()</function> variants.</para>
+
+    <para><function>sd_varlink_invoke_and_upgrade()</function> is the asynchronous counterpart of
+    <function>sd_varlink_call_and_upgrade()</function>, for connections attached to an event loop with
+    <citerefentry><refentrytitle>sd_varlink_attach_event</refentrytitle><manvolnum>3</manvolnum></citerefentry>.
+    It enqueues the method call with the upgrade request and returns immediately. Once the reply arrives it
+    is passed to the reply callback installed with
+    <citerefentry><refentrytitle>sd_varlink_bind_reply</refentrytitle><manvolnum>3</manvolnum></citerefentry>.
+    If the server replied with a Varlink error, the connection stays in the Varlink protocol and may be used
+    further. Otherwise the connection is upgraded on the next dispatch after the reply callback returned
+    (i.e. the next invocation of <function>sd_varlink_process()</function>, typically by the event loop):
+    it is detached from the event loop, disconnected from the Varlink
+    protocol, and the underlying file descriptors are passed to the upgrade callback installed with
+    <citerefentry><refentrytitle>sd_varlink_bind_upgrade</refentrytitle><manvolnum>3</manvolnum></citerefentry>,
+    with the same semantics as the descriptors returned by <function>sd_varlink_call_and_upgrade()</function>.
+    The callback takes ownership of both descriptors. Unlike with regular method calls, no further method
+    calls may be enqueued on the connection while an upgrade request is pending, since the connection leaves
+    the Varlink protocol once the request has been answered. <function>sd_varlink_invoke_and_upgradeb()</function> and
+    <function>sd_varlink_invoke_and_upgradebo()</function> relate to <function>sd_varlink_invoke_and_upgrade()</function>
+    like the <function>…b()</function> and <function>…bo()</function> variants described above.</para>
   </refsect1>
 
   <refsect1>
     <title>Return Value</title>
 
-    <para>On success, these functions return a positive integer if the connection was upgraded, or zero if the
-    server returned a Varlink error and <parameter>ret_error_id</parameter> was set to it. On failure, they
-    return a negative errno-style error code.</para>
+    <para>On success, <function>sd_varlink_call_and_upgrade()</function> and its variants return a positive
+    integer if the connection was upgraded. If the server returns a Varlink error and
+    <parameter>ret_error_id</parameter> was provided by the caller, it is set to the error identifier and zero
+    is returned. Otherwise, the Varlink error is treated as a failure.
+    <function>sd_varlink_invoke_and_upgrade()</function> and its variants return zero once the request has
+    been enqueued. On failure, all functions return a negative errno-style error code.</para>
 
     <refsect2>
       <title>Errors</title>
@@ -138,6 +184,16 @@
           <listitem><para>The Varlink connection object is not connected.</para></listitem>
         </varlistentry>
 
+        <varlistentry>
+          <term><constant>-EBUSY</constant></term>
+
+          <listitem><para>Another method call is pending on the connection. Note that while a request
+          enqueued with <function>sd_varlink_invoke_and_upgrade()</function> is pending, this error is also
+          returned by any other function that enqueues a method call on the connection, such as
+          <citerefentry><refentrytitle>sd_varlink_call</refentrytitle><manvolnum>3</manvolnum></citerefentry>
+          or <function>sd_varlink_invoke()</function>.</para></listitem>
+        </varlistentry>
+
         <varlistentry>
           <term><constant>-EPROTO</constant></term>
 
@@ -161,6 +217,9 @@
     <para><function>sd_varlink_call_and_upgrade()</function> was added in version 261.</para>
     <para><function>sd_varlink_call_and_upgradeb()</function> and
     <function>sd_varlink_call_and_upgradebo()</function> were added in version 262.</para>
+    <para><function>sd_varlink_invoke_and_upgrade()</function>,
+    <function>sd_varlink_invoke_and_upgradeb()</function> and
+    <function>sd_varlink_invoke_and_upgradebo()</function> were added in version 263.</para>
   </refsect1>
 
   <refsect1>

src/libsystemd/libsystemd.sym

@@ -1117,4 +1117,6 @@ LIBSYSTEMD_263 {
 global:
         sd_device_new_from_device_node_fd;
         sd_device_trigger_with_args;
+        sd_varlink_invoke_and_upgrade;
+        sd_varlink_invoke_and_upgradeb;
 } LIBSYSTEMD_262;

src/libsystemd/sd-varlink/sd-varlink.c

@@ -51,6 +51,7 @@ static const char* const varlink_state_table[_VARLINK_STATE_MAX] = {
         [VARLINK_IDLE_CLIENT]               = "idle-client",
         [VARLINK_AWAITING_REPLY]            = "awaiting-reply",
         [VARLINK_AWAITING_REPLY_MORE]       = "awaiting-reply-more",
+        [VARLINK_AWAITING_REPLY_UPGRADE]    = "awaiting-reply-upgrade",
         [VARLINK_CALLING]                   = "calling",
         [VARLINK_CALLED]                    = "called",
         [VARLINK_COLLECTING]                = "collecting",
@@ -104,7 +105,7 @@ static JsonStreamPhase varlink_phase(void *userdata) {
 
         /* Client side reading a reply with the per-call deadline in force. */
         if (IN_SET(v->state,
-                   VARLINK_AWAITING_REPLY, VARLINK_AWAITING_REPLY_MORE,
+                   VARLINK_AWAITING_REPLY, VARLINK_AWAITING_REPLY_MORE, VARLINK_AWAITING_REPLY_UPGRADE,
                    VARLINK_CALLING, VARLINK_COLLECTING) &&
             !v->current)
                 return JSON_STREAM_PHASE_AWAITING_REPLY;
@@ -736,7 +737,7 @@ static int varlink_write(sd_varlink *v) {
 static int varlink_read(sd_varlink *v) {
         assert(v);
 
-        if (!IN_SET(v->state, VARLINK_AWAITING_REPLY, VARLINK_AWAITING_REPLY_MORE, VARLINK_CALLING, VARLINK_COLLECTING, VARLINK_IDLE_SERVER))
+        if (!IN_SET(v->state, VARLINK_AWAITING_REPLY, VARLINK_AWAITING_REPLY_MORE, VARLINK_AWAITING_REPLY_UPGRADE, VARLINK_CALLING, VARLINK_COLLECTING, VARLINK_IDLE_SERVER))
                 return 0;
         if (v->current)
                 return 0;
@@ -861,7 +862,7 @@ static int varlink_dispatch_reply(sd_varlink *v) {
 
         assert(v);
 
-        if (!IN_SET(v->state, VARLINK_AWAITING_REPLY, VARLINK_AWAITING_REPLY_MORE, VARLINK_CALLING, VARLINK_COLLECTING))
+        if (!IN_SET(v->state, VARLINK_AWAITING_REPLY, VARLINK_AWAITING_REPLY_MORE, VARLINK_AWAITING_REPLY_UPGRADE, VARLINK_CALLING, VARLINK_COLLECTING))
                 return 0;
         if (!v->current)
                 return 0;
@@ -917,7 +918,9 @@ static int varlink_dispatch_reply(sd_varlink *v) {
 
         v->current_reply_flags = flags;
 
-        if (IN_SET(v->state, VARLINK_AWAITING_REPLY, VARLINK_AWAITING_REPLY_MORE)) {
+        if (IN_SET(v->state, VARLINK_AWAITING_REPLY, VARLINK_AWAITING_REPLY_MORE, VARLINK_AWAITING_REPLY_UPGRADE)) {
+                bool upgrade = v->state == VARLINK_AWAITING_REPLY_UPGRADE;
+
                 varlink_set_state(v, VARLINK_PROCESSING_REPLY);
 
                 if (v->reply_callback) {
@@ -931,6 +934,24 @@ static int varlink_dispatch_reply(sd_varlink *v) {
                 if (v->state == VARLINK_PROCESSING_REPLY) {
                         assert(v->n_pending > 0);
 
+                        if (upgrade) {
+                                assert(v->n_pending == 1);
+                                v->n_pending--;
+
+                                if (error) {
+                                        /* The server refused the upgrade, hence the connection stays in the
+                                         * Varlink protocol and can be used further. */
+                                        json_stream_set_flags(&v->stream, JSON_STREAM_BOUNDED_READS, false);
+                                        varlink_set_state(v, VARLINK_IDLE_CLIENT);
+                                } else
+                                        /* The server accepted the upgrade. The fds are handed to the upgrade
+                                         * callback on the next dispatch, see varlink_dispatch_upgrade(), so
+                                         * that we don't invoke two callbacks in one go. */
+                                        varlink_set_state(v, VARLINK_UPGRADING);
+
+                                return 1;
+                        }
+
                         if (!FLAGS_SET(flags, SD_VARLINK_REPLY_CONTINUES))
                                 v->n_pending--;
 
@@ -1505,6 +1526,10 @@ static int varlink_dispatch_upgrade(sd_varlink *v) {
 
         assert(v);
 
+        /* Completes a protocol upgrade, both for servers (after sd_varlink_respond_and_upgrade()) and
+         * clients (after the reply to sd_varlink_invoke_and_upgrade() arrived): hands the connection's fds
+         * to the upgrade callback, once everything has been written out. */
+
         if (v->state != VARLINK_UPGRADING)
                 return 0;
 
@@ -1963,6 +1988,64 @@ _public_ int sd_varlink_invokeb(sd_varlink *v, const char *method, ...) {
         return sd_varlink_invoke(v, method, parameters);
 }
 
+_public_ int sd_varlink_invoke_and_upgrade(sd_varlink *v, const char *method, sd_json_variant *parameters) {
+        _cleanup_(sd_json_variant_unrefp) sd_json_variant *m = NULL;
+        int r;
+
+        assert_return(v, -EINVAL);
+        assert_return(method, -EINVAL);
+
+        if (v->state == VARLINK_DISCONNECTED)
+                return varlink_log_errno(v, SYNTHETIC_ERRNO(ENOTCONN), "Not connected.");
+
+        /* Unlike for regular method calls we do not allow pipelining here: once the reply arrives the
+         * connection leaves the Varlink protocol, hence no other calls may be pending. */
+        if (v->state != VARLINK_IDLE_CLIENT)
+                return varlink_log_errno(v, SYNTHETIC_ERRNO(EBUSY), "Connection busy.");
+
+        assert(v->n_pending == 0); /* n_pending can't be > 0 if we are in VARLINK_IDLE_CLIENT state */
+
+        r = sd_json_buildo(
+                        &m,
+                        SD_JSON_BUILD_PAIR_STRING("method", method),
+                        JSON_BUILD_PAIR_VARIANT_NON_EMPTY("parameters", parameters),
+                        SD_JSON_BUILD_PAIR_BOOLEAN("upgrade", true));
+        if (r < 0)
+                return varlink_log_errno(v, r, "Failed to build json message: %m");
+
+        /* Make sure we don't consume any data of the upgraded protocol that might follow the reply */
+        json_stream_set_flags(&v->stream, JSON_STREAM_BOUNDED_READS, true);
+
+        r = varlink_enqueue(v, m);
+        if (r < 0) {
+                json_stream_set_flags(&v->stream, JSON_STREAM_BOUNDED_READS, false);
+                return varlink_log_errno(v, r, "Failed to enqueue json message: %m");
+        }
+
+        varlink_set_state(v, VARLINK_AWAITING_REPLY_UPGRADE);
+        v->n_pending++;
+        json_stream_mark_activity(&v->stream);
+
+        return 0;
+}
+
+_public_ int sd_varlink_invoke_and_upgradeb(sd_varlink *v, const char *method, ...) {
+        _cleanup_(sd_json_variant_unrefp) sd_json_variant *parameters = NULL;
+        va_list ap;
+        int r;
+
+        assert_return(v, -EINVAL);
+        assert_return(method, -EINVAL);
+
+        va_start(ap, method);
+        r = sd_json_buildv(&parameters, ap);
+        va_end(ap);
+        if (r < 0)
+                return varlink_log_errno(v, r, "Failed to build json message: %m");
+
+        return sd_varlink_invoke_and_upgrade(v, method, parameters);
+}
+
 _public_ int sd_varlink_observe(sd_varlink *v, const char *method, sd_json_variant *parameters) {
         _cleanup_(sd_json_variant_unrefp) sd_json_variant *m = NULL;
         int r;

src/libsystemd/sd-varlink/test-varlink.c

@@ -1282,6 +1282,142 @@ TEST(upgrade) {
         ASSERT_OK(sd_future_result(f));
 }
 
+typedef struct AsyncUpgradeTest {
+        sd_event *event;
+        bool got_reply, got_upgrade, echo_done, got_error;
+        int input_fd, output_fd;
+} AsyncUpgradeTest;
+
+static int async_upgrade_echo_fiber(void *arg) {
+        AsyncUpgradeTest *t = ASSERT_PTR(arg);
+
+        /* Same exchange as in upgrade_client_fiber(), over the fds donated to the upgrade callback */
+        static const char msg[] = "Hello!";
+        ASSERT_OK(loop_write(t->output_fd, msg, strlen(msg)));
+        ASSERT_OK_ERRNO(shutdown(t->output_fd, SHUT_WR));
+
+        char buf[64] = {};
+        ssize_t n = ASSERT_OK(loop_read(t->input_fd, buf, strlen(msg), /* do_poll= */ true));
+        ASSERT_EQ((size_t) n, strlen(msg));
+        ASSERT_STREQ(buf, "!olleH");
+
+        t->input_fd = safe_close(t->input_fd);
+        t->output_fd = safe_close(t->output_fd);
+        t->echo_done = true;
+
+        ASSERT_OK(sd_event_exit(t->event, EXIT_SUCCESS));
+        return 0;
+}
+
+static int async_upgrade_on_reply(sd_varlink *link, sd_json_variant *parameters, const char *error_id, sd_varlink_reply_flags_t flags, void *userdata) {
+        AsyncUpgradeTest *t = ASSERT_PTR(userdata);
+
+        ASSERT_FALSE(t->got_upgrade); /* the reply callback must run before the upgrade callback */
+        ASSERT_NULL(error_id);
+        ASSERT_FALSE(FLAGS_SET(flags, SD_VARLINK_REPLY_ERROR));
+        t->got_reply = true;
+        return 0;
+}
+
+static int async_upgrade_on_upgrade(sd_varlink *link, int input_fd, int output_fd, void *userdata) {
+        AsyncUpgradeTest *t = ASSERT_PTR(userdata);
+
+        ASSERT_TRUE(t->got_reply);
+        ASSERT_GE(input_fd, 0);
+        ASSERT_GE(output_fd, 0);
+        ASSERT_NE(input_fd, output_fd);
+        ASSERT_FALSE(sd_varlink_is_connected(link));
+
+        t->got_upgrade = true;
+        t->input_fd = input_fd;
+        t->output_fd = output_fd;
+
+        /* The raw I/O blocks, hence do it in a fiber so that the server can make progress */
+        _cleanup_(sd_future_unrefp) sd_future *f = NULL;
+        ASSERT_OK(sd_fiber_new(t->event, "async-upgrade-echo", async_upgrade_echo_fiber, t, &f));
+        ASSERT_OK(sd_fiber_set_floating(f, true));
+        return 0;
+}
+
+static int async_upgrade_on_error_reply(sd_varlink *link, sd_json_variant *parameters, const char *error_id, sd_varlink_reply_flags_t flags, void *userdata) {
+        AsyncUpgradeTest *t = ASSERT_PTR(userdata);
+
+        ASSERT_STREQ(error_id, SD_VARLINK_ERROR_METHOD_NOT_FOUND);
+        ASSERT_TRUE(FLAGS_SET(flags, SD_VARLINK_REPLY_ERROR));
+        t->got_error = true;
+        return 0;
+}
+
+static int async_upgrade_on_unexpected_upgrade(sd_varlink *link, int input_fd, int output_fd, void *userdata) {
+        safe_close(input_fd);
+        safe_close(output_fd);
+        assert_not_reached();
+}
+
+TEST(upgrade_async) {
+        _cleanup_(sd_varlink_server_unrefp) sd_varlink_server *s = NULL;
+        _cleanup_(rm_rf_physical_and_freep) char *tmpdir = NULL;
+        _cleanup_(sd_event_unrefp) sd_event *e = NULL;
+        const char *sp;
+
+        ASSERT_OK(mkdtemp_malloc("/tmp/varlink-test-XXXXXX", &tmpdir));
+        sp = strjoina(tmpdir, "/socket");
+
+        ASSERT_OK(sd_event_new(&e));
+
+        ASSERT_OK(sd_varlink_server_new(&s, SD_VARLINK_SERVER_UPGRADABLE));
+        ASSERT_OK(sd_varlink_server_set_description(s, "upgrade-server"));
+        ASSERT_OK(varlink_server_bind_fiber(s, "io.test.Upgrade", method_upgrade));
+        ASSERT_OK(sd_varlink_server_listen_address(s, sp, 0600));
+        ASSERT_OK(sd_varlink_server_attach_event(s, e, 0));
+
+        /* First a client whose upgrade request is answered with an error: the connection must stay usable */
+        AsyncUpgradeTest t2 = { .event = e, .input_fd = -EBADF, .output_fd = -EBADF };
+        _cleanup_(sd_varlink_unrefp) sd_varlink *c2 = NULL;
+        ASSERT_OK(sd_varlink_connect_address(&c2, sp));
+        ASSERT_OK(sd_varlink_set_description(c2, "async-upgrade-error-client"));
+        ASSERT_OK(sd_varlink_attach_event(c2, e, 0));
+        sd_varlink_set_userdata(c2, &t2);
+        ASSERT_OK(sd_varlink_bind_reply(c2, async_upgrade_on_error_reply));
+        ASSERT_OK(sd_varlink_bind_upgrade(c2, async_upgrade_on_unexpected_upgrade));
+        ASSERT_OK(sd_varlink_invoke_and_upgrade(c2, "io.test.NoSuchMethod", /* parameters= */ NULL));
+
+        while (!t2.got_error)
+                ASSERT_OK(sd_event_run(e, UINT64_MAX));
+
+        ASSERT_TRUE(sd_varlink_is_connected(c2));
+        ASSERT_OK_POSITIVE(sd_varlink_is_idle(c2));
+
+        /* After the refused upgrade regular calls work again */
+        t2.got_error = false;
+        ASSERT_OK(sd_varlink_invoke(c2, "io.test.NoSuchMethod", /* parameters= */ NULL));
+        while (!t2.got_error)
+                ASSERT_OK(sd_event_run(e, UINT64_MAX));
+        ASSERT_TRUE(sd_varlink_is_connected(c2));
+
+        /* Now a client driven entirely by the event loop whose upgrade is accepted */
+        AsyncUpgradeTest t = { .event = e, .input_fd = -EBADF, .output_fd = -EBADF };
+        _cleanup_(sd_varlink_unrefp) sd_varlink *c = NULL;
+        ASSERT_OK(sd_varlink_connect_address(&c, sp));
+        ASSERT_OK(sd_varlink_set_description(c, "async-upgrade-client"));
+        ASSERT_OK(sd_varlink_attach_event(c, e, 0));
+        sd_varlink_set_userdata(c, &t);
+        ASSERT_OK(sd_varlink_bind_reply(c, async_upgrade_on_reply));
+        ASSERT_OK(sd_varlink_bind_upgrade(c, async_upgrade_on_upgrade));
+        ASSERT_OK(sd_varlink_invoke_and_upgrade(c, "io.test.Upgrade", /* parameters= */ NULL));
+
+        /* No other calls may be pipelined behind an upgrade request */
+        ASSERT_ERROR(sd_varlink_invoke(c, "io.test.Upgrade", /* parameters= */ NULL), EBUSY);
+        ASSERT_ERROR(sd_varlink_invoke_and_upgrade(c, "io.test.Upgrade", /* parameters= */ NULL), EBUSY);
+
+        ASSERT_OK(sd_event_loop(e));
+
+        ASSERT_TRUE(t.got_reply);
+        ASSERT_TRUE(t.got_upgrade);
+        ASSERT_TRUE(t.echo_done);
+        ASSERT_FALSE(sd_varlink_is_connected(c));
+}
+
 static int upgrade_pipelining_client_fiber(void *arg) {
         union sockaddr_union sa = {};
         _cleanup_close_ int fd = -EBADF;

src/libsystemd/sd-varlink/varlink-internal.h

@@ -15,6 +15,7 @@ typedef enum VarlinkState {
         VARLINK_IDLE_CLIENT,
         VARLINK_AWAITING_REPLY,
         VARLINK_AWAITING_REPLY_MORE,
+        VARLINK_AWAITING_REPLY_UPGRADE,
         VARLINK_CALLING,
         VARLINK_CALLED,
         VARLINK_COLLECTING,
@@ -55,6 +56,7 @@ typedef enum VarlinkState {
                VARLINK_IDLE_CLIENT,                     \
                VARLINK_AWAITING_REPLY,                  \
                VARLINK_AWAITING_REPLY_MORE,             \
+               VARLINK_AWAITING_REPLY_UPGRADE,          \
                VARLINK_CALLING,                         \
                VARLINK_CALLED,                          \
                VARLINK_COLLECTING,                      \

src/systemd/sd-varlink.h

@@ -162,6 +162,15 @@ int sd_varlink_invokeb(sd_varlink *v, const char *method, ...);
 #define sd_varlink_invokebo(v, method, ...)                                \
         sd_varlink_invokeb((v), (method), SD_JSON_BUILD_OBJECT(__VA_ARGS__))
 
+/* Enqueue method call with upgrade request. The reply is delivered to the reply callback. If the server
+ * accepted the upgrade, the connection is disconnected from the Varlink protocol afterwards and its fds are
+ * handed to the upgrade callback, see sd_varlink_bind_upgrade(). The asynchronous counterpart of
+ * sd_varlink_call_and_upgrade(). */
+int sd_varlink_invoke_and_upgrade(sd_varlink *v, const char *method, sd_json_variant *parameters);
+int sd_varlink_invoke_and_upgradeb(sd_varlink *v, const char *method, ...);
+#define sd_varlink_invoke_and_upgradebo(v, method, ...)                    \
+        sd_varlink_invoke_and_upgradeb((v), (method), SD_JSON_BUILD_OBJECT(__VA_ARGS__))
+
 /* Enqueue method call, expect a reply now, and possibly more later, which are all delivered to the reply callback */
 int sd_varlink_observe(sd_varlink *v, const char *method, sd_json_variant *parameters);
 int sd_varlink_observeb(sd_varlink *v, const char *method, ...);