sd-varlink: add sd_varlink_wants_reply()

Method handlers can call helpers that send a reply before returning an
error. Callers then need to know which helper failures have already
replied before trying to send their own error.

Let's add sd_varlink_wants_reply() so callers can check whether an
incoming method call still needs a reply. The check also covers calls
deferred to a callback or fiber.

man/rules/meson.build

@@ -991,7 +991,10 @@ manpages = [
    'sd_varlink_connect_fd_pair',
    'sd_varlink_connect_url'],
   ''],
- ['sd_varlink_is_connected', '3', ['sd_varlink_is_idle'], ''],
+ ['sd_varlink_is_connected',
+  '3',
+  ['sd_varlink_is_idle', 'sd_varlink_wants_reply'],
+  ''],
  ['sd_varlink_push_fd', '3', ['sd_varlink_push_dup_fd'], ''],
  ['sd_varlink_reply_and_upgrade',
   '3',

man/sd_varlink_is_connected.xml

@@ -18,8 +18,10 @@
   <refnamediv>
     <refname>sd_varlink_is_connected</refname>
     <refname>sd_varlink_is_idle</refname>
+    <refname>sd_varlink_wants_reply</refname>
 
-    <refpurpose>Query if a Varlink connection object is currently connected or idle</refpurpose>
+    <refpurpose>Query if a Varlink connection object is currently connected, idle, or expected to reply to
+    a method call</refpurpose>
   </refnamediv>
 
   <refsynopsisdiv>
@@ -35,6 +37,11 @@
         <funcdef>int <function>sd_varlink_is_idle</function></funcdef>
         <paramdef>sd_varlink *<parameter>link</parameter></paramdef>
       </funcprototype>
+
+      <funcprototype>
+        <funcdef>int <function>sd_varlink_wants_reply</function></funcdef>
+        <paramdef>sd_varlink *<parameter>link</parameter></paramdef>
+      </funcprototype>
     </funcsynopsis>
   </refsynopsisdiv>
 
@@ -50,16 +57,42 @@
     currently connected but idle, i.e. may accept a method call for enqueuing (if client side of a Varlink
     connection) or is waiting for more incoming method calls to arrive (if server side of a Varlink
     connection).</para>
+
+    <para><function>sd_varlink_wants_reply()</function> checks whether the specified Varlink connection
+    object has an incoming method call awaiting a reply. No final reply or error has been enqueued for
+    the call yet, and the method call was not issued with the <literal>oneway</literal> flag set. This is
+    useful on the server side of a Varlink connection, in particular in method call handlers that invoke
+    helper functions that might or might not have replied
+    to the method call already, for example with an error, before returning. Once a final reply or error
+    has been enqueued for transmission, this function returns 0.</para>
+
+    <para>While a method call invoked with the <literal>more</literal> flag set is processed, this function
+    continues to return a positive value after <function>sd_varlink_notify()</function> has been called,
+    as a final reply is still expected. If <function>sd_varlink_set_sentinel()</function> has been called,
+    <function>sd_varlink_reply()</function> buffers the latest reply until the method handler returns so
+    that the reply can be flagged as non-final if further replies follow.
+    <function>sd_varlink_wants_reply()</function> continues to return a positive value until the buffered
+    reply is enqueued as the final reply.</para>
+
+    <para>A positive value is also returned while a fiber method handler runs, or after a method handler
+    returns without replying, having taken a reference to the connection object to reply asynchronously
+    later.</para>
+
+    <para>On the client side of a Varlink connection this function always returns 0. If <parameter>link</parameter> is
+    <constant>NULL</constant>, the function returns 0, since a non-existent connection does not expect a
+    reply.</para>
   </refsect1>
 
   <refsect1>
     <title>Return Value</title>
 
     <para>If the connection is currently connected <function>sd_varlink_is_connected()</function> returns a
-    positive non-zero value. If disconnected (or if <constant>NULL</constant> is specified) it returns
-    0. If the connection currently is idle
-    <function>sd_varlink_is_idle()</function> returns a positive non-zero integer. If not, returns 0. On
-    failure, <function>sd_varlink_is_idle()</function> returns a negative <varname>errno</varname> error
+    positive non-zero value. If disconnected (or if <constant>NULL</constant> is specified) it returns 0.
+    If the connection currently is idle <function>sd_varlink_is_idle()</function> returns a positive
+    non-zero integer. If not, returns 0. If a reply to the method call currently being processed is
+    expected <function>sd_varlink_wants_reply()</function> returns a positive non-zero integer. If not (or
+    if <constant>NULL</constant> is specified), it returns 0. On failure,
+    <function>sd_varlink_is_idle()</function> returns a negative <varname>errno</varname> error
     code.</para>
 
     <refsect2>
@@ -72,7 +105,8 @@
           <term><constant>-EINVAL</constant></term>
 
           <listitem><para>The specified Varlink connection is invalid. (Not returned by
-          <function>sd_varlink_is_connected()</function>.)</para></listitem>
+          <function>sd_varlink_is_connected()</function> and
+          <function>sd_varlink_wants_reply()</function>.)</para></listitem>
         </varlistentry>
       </variablelist>
     </refsect2>
@@ -84,6 +118,7 @@
     <title>History</title>
     <para><function>sd_varlink_is_connected()</function> was added in version 259.</para>
     <para><function>sd_varlink_is_idle()</function> was added in version 257.</para>
+    <para><function>sd_varlink_wants_reply()</function> was added in version 263.</para>
   </refsect1>
 
   <refsect1>
@@ -92,6 +127,7 @@
     <para><simplelist type="inline">
       <member><citerefentry><refentrytitle>systemd</refentrytitle><manvolnum>1</manvolnum></citerefentry></member>
       <member><citerefentry><refentrytitle>sd-varlink</refentrytitle><manvolnum>3</manvolnum></citerefentry></member>
+      <member><citerefentry><refentrytitle>sd_varlink_send</refentrytitle><manvolnum>3</manvolnum></citerefentry></member>
     </simplelist></para>
   </refsect1>
 

src/libsystemd/libsystemd.sym

@@ -1120,4 +1120,5 @@ global:
         sd_journal_get_data_to_fd;
         sd_varlink_invoke_and_upgrade;
         sd_varlink_invoke_and_upgradeb;
+        sd_varlink_wants_reply;
 } LIBSYSTEMD_262;

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

@@ -1736,6 +1736,14 @@ _public_ int sd_varlink_is_connected(sd_varlink *v) {
         return v && v->state != VARLINK_DISCONNECTED;
 }
 
+_public_ int sd_varlink_wants_reply(sd_varlink *v) {
+        return v && (VARLINK_STATE_WANTS_REPLY(v->state) ||
+                     IN_SET(v->state,
+                            VARLINK_PENDING_METHOD,
+                            VARLINK_PENDING_METHOD_MORE,
+                            VARLINK_PENDING_METHOD_UPGRADE));
+}
+
 _public_ int sd_varlink_get_fd(sd_varlink *v) {
         assert_return(v, -EINVAL);
 

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

@@ -650,12 +650,18 @@ static int method_with_oneway_sentinel(sd_varlink *link, sd_json_variant *parame
         /* The method was called oneway, so sd_varlink_set_sentinel() should be a no-op and the server should
          * transition back to idle without sending any reply. */
         ASSERT_TRUE(FLAGS_SET(flags, SD_VARLINK_METHOD_ONEWAY));
+        ASSERT_OK_ZERO(sd_varlink_wants_reply(link));
         ASSERT_OK(sd_varlink_set_sentinel(link, "io.test.SentinelError"));
+        ASSERT_OK_ZERO(sd_varlink_wants_reply(link));
         return 0;
 }
 
 static int method_oneway_sentinel_pong(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata) {
-        return sd_varlink_replybo(link, SD_JSON_BUILD_PAIR_STRING("result", "pong"));
+        /* A regular method call expects a reply, until we sent one */
+        ASSERT_OK_POSITIVE(sd_varlink_wants_reply(link));
+        ASSERT_OK(sd_varlink_replybo(link, SD_JSON_BUILD_PAIR_STRING("result", "pong")));
+        ASSERT_OK_ZERO(sd_varlink_wants_reply(link));
+        return 0;
 }
 
 static int reply_oneway_sentinel_pong(sd_varlink *link, sd_json_variant *parameters, const char *error_id, sd_varlink_reply_flags_t flags, void *userdata) {
@@ -663,6 +669,7 @@ static int reply_oneway_sentinel_pong(sd_varlink *link, sd_json_variant *paramet
          * properly handled a subsequent regular method call. */
         ASSERT_NULL(error_id);
         ASSERT_STREQ(sd_json_variant_string(sd_json_variant_by_key(parameters, "result")), "pong");
+        ASSERT_OK_ZERO(sd_varlink_wants_reply(link)); /* Client side never wants to reply */
         ASSERT_OK(sd_event_exit(sd_varlink_get_event(link), EXIT_SUCCESS));
         return 0;
 }
@@ -877,6 +884,7 @@ static int method_fiber_sentinel_error(sd_varlink *link, sd_json_variant *parame
         /* Set an error sentinel from a fiber callback and return without sending a reply. The sentinel
          * error should still be propagated by the fiber's post-callback logic, even though the varlink
          * state has already been transitioned to VARLINK_PENDING_METHOD by the time the fiber runs. */
+        ASSERT_OK_POSITIVE(sd_varlink_wants_reply(link));
         ASSERT_OK(sd_varlink_set_sentinel(link, "io.test.SentinelError"));
         return 0;
 }
@@ -991,7 +999,9 @@ static int fiber_stashed_deferred_reply(sd_event_source *s, void *userdata) {
         _cleanup_(sd_varlink_unrefp) sd_varlink *link = ASSERT_PTR(userdata);
 
         sd_event_source_disable_unref(s);
+        ASSERT_OK_POSITIVE(sd_varlink_wants_reply(link));
         ASSERT_OK(sd_varlink_replybo(link, SD_JSON_BUILD_PAIR_STRING("result", "stashed")));
+        ASSERT_OK_ZERO(sd_varlink_wants_reply(link));
         return 0;
 }
 
@@ -1001,6 +1011,7 @@ static int method_fiber_stash(sd_varlink *link, sd_json_variant *parameters, sd_
          * this as a valid deferred-reply case instead of failing the connection. */
         sd_event_source *source;
 
+        ASSERT_OK_POSITIVE(sd_varlink_wants_reply(link));
         ASSERT_OK(sd_event_add_defer(sd_varlink_get_event(link), &source, fiber_stashed_deferred_reply, sd_varlink_ref(link)));
         ASSERT_OK(sd_event_source_set_enabled(source, SD_EVENT_ONESHOT));
         return 0;
@@ -1044,6 +1055,7 @@ static int method_with_fd_sentinel(sd_varlink *link, sd_json_variant *parameters
         _cleanup_close_ int fd1 = -EBADF, fd2 = -EBADF;
 
         ASSERT_TRUE(FLAGS_SET(flags, SD_VARLINK_METHOD_MORE));
+        ASSERT_OK_POSITIVE(sd_varlink_wants_reply(link));
 
         /* Set a sentinel so sd_varlink_reply() defers sending: each reply and its pushed fds are captured in
          * the queue, and the last one is sent as the final reply when the callback returns. */
@@ -1054,12 +1066,14 @@ static int method_with_fd_sentinel(sd_varlink *link, sd_json_variant *parameters
         ASSERT_OK_EQ(sd_varlink_push_fd(link, fd1), 0);
         TAKE_FD(fd1);
         ASSERT_OK(sd_varlink_replybo(link, SD_JSON_BUILD_PAIR_INTEGER("index", 0)));
+        ASSERT_OK_POSITIVE(sd_varlink_wants_reply(link));
 
         /* Second reply: push one fd with "beta" content */
         ASSERT_OK(fd2 = memfd_new_and_seal_string("data", "beta"));
         ASSERT_OK_EQ(sd_varlink_push_fd(link, fd2), 0);
         TAKE_FD(fd2);
         ASSERT_OK(sd_varlink_replybo(link, SD_JSON_BUILD_PAIR_INTEGER("index", 1)));
+        ASSERT_OK_POSITIVE(sd_varlink_wants_reply(link));
 
         return 0;
 }
@@ -1094,7 +1108,7 @@ static int reply_sentinel_fd(sd_varlink *link, sd_json_variant *parameters, cons
         return 0;
 }
 
-TEST(sentinel_with_fds) {
+static void run_sentinel_with_fds(bool use_fiber) {
         _cleanup_(sd_event_unrefp) sd_event *e = NULL;
         ASSERT_OK(sd_event_default(&e));
 
@@ -1103,7 +1117,10 @@ TEST(sentinel_with_fds) {
 
         ASSERT_OK(sd_varlink_server_attach_event(s, e, /* priority= */ 0));
 
-        ASSERT_OK(sd_varlink_server_bind_method(s, "io.test.FDSentinel", method_with_fd_sentinel));
+        if (use_fiber)
+                ASSERT_OK(varlink_server_bind_fiber(s, "io.test.FDSentinel", method_with_fd_sentinel));
+        else
+                ASSERT_OK(sd_varlink_server_bind_method(s, "io.test.FDSentinel", method_with_fd_sentinel));
 
         int connfd[2];
         ASSERT_OK_ERRNO(socketpair(AF_UNIX, SOCK_STREAM|SOCK_NONBLOCK|SOCK_CLOEXEC, 0, connfd));
@@ -1125,10 +1142,22 @@ TEST(sentinel_with_fds) {
         ASSERT_OK(sd_event_loop(e));
 }
 
+TEST(sentinel_with_fds) {
+        run_sentinel_with_fds(/* use_fiber= */ false);
+}
+
+TEST(fiber_sentinel_with_fds) {
+        run_sentinel_with_fds(/* use_fiber= */ true);
+}
+
 static int method_with_notify_then_error(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata) {
         /* Send a notify first, then return an error. The notify should be received before the error. */
         ASSERT_OK(sd_varlink_notifybo(link, SD_JSON_BUILD_PAIR_STRING("status", "in-progress")));
-        return sd_varlink_error(link, "io.test.OperationFailed", /* parameters= */ NULL);
+        ASSERT_OK_POSITIVE(sd_varlink_wants_reply(link));
+
+        int r = sd_varlink_error(link, "io.test.OperationFailed", /* parameters= */ NULL);
+        ASSERT_OK_ZERO(sd_varlink_wants_reply(link));
+        return r;
 }
 
 static int reply_notify_then_error(sd_varlink *link, sd_json_variant *parameters, const char *error_id, sd_varlink_reply_flags_t flags, void *userdata) {
@@ -1185,11 +1214,14 @@ static int method_upgrade(sd_varlink *link, sd_json_variant *parameters, sd_varl
         int r;
 
         ASSERT_TRUE(FLAGS_SET(flags, SD_VARLINK_METHOD_UPGRADE));
+        ASSERT_OK_POSITIVE(sd_varlink_wants_reply(link));
 
         r = sd_varlink_reply_and_upgrade(link, /* parameters= */ NULL, &input_fd, &output_fd);
         if (r < 0)
                 return r;
 
+        ASSERT_OK_ZERO(sd_varlink_wants_reply(link));
+
         /* After upgrade, do raw I/O: read until the client shuts down its write side (giving us a clean
          * EOF), reverse what we got, and write it back. Use suspending I/O so other fibers (the client)
          * can make progress while we're waiting on the socket. */
@@ -1870,6 +1902,11 @@ TEST(is_connected_null) {
         ASSERT_OK_ZERO(sd_varlink_is_connected(NULL));
 }
 
+TEST(wants_reply_null) {
+        /* A non-existent connection wants no reply */
+        ASSERT_OK_ZERO(sd_varlink_wants_reply(NULL));
+}
+
 TEST(add_connection_failure_cleanup) {
         _cleanup_(sd_varlink_server_unrefp) sd_varlink_server *s = NULL;
         _cleanup_close_pair_ int fds[2] = EBADF_PAIR;

src/systemd/sd-varlink.h

@@ -113,6 +113,7 @@ int sd_varlink_wait(sd_varlink *v, uint64_t timeout);
 
 int sd_varlink_is_idle(sd_varlink *v);
 int sd_varlink_is_connected(sd_varlink *v);
+int sd_varlink_wants_reply(sd_varlink *v);
 
 int sd_varlink_flush(sd_varlink *v);
 int sd_varlink_close(sd_varlink *v);