sd-varlink: Introduce varlink_set_sentinel()

Streaming methods which are not used as a continuous subscription but
instead only send a series of objects all end up with the same workaround
to be able to figure out when to send sd_varlink_reply() or sd_varlink_notify().
Let's generalize this in sd-varlink itself.

Let's introduce the concept of a sentinel, which is a reply that will be sent
by sd-varlink if no other reply was queued by a method callback. The sentinel
is configured with varlink_set_sentinel(). If a sentinel is configured,
sd_varlink_reply() can be used more than once in streaming methods to queue
multiple values to stream to the client. The last queued reply is not sent
until the callback finishes. When the callback finishes, the last reply is
sent without "continues: more". If no reply was queued, the sentinel is sent.

This always using only sd_varlink_reply() in such streaming methods and
leaves sd_varlink_notify() available solely for continuous subscription
streaming methods, where we never use sd_varlink_reply() and instead disconnect
when the server exits.

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

@@ -616,6 +616,12 @@ static void varlink_clear_current(sd_varlink *v) {
         close_many(v->input_fds, v->n_input_fds);
         v->input_fds = mfree(v->input_fds);
         v->n_input_fds = 0;
+
+        v->previous = varlink_json_queue_item_free(v->previous);
+        if (v->sentinel != POINTER_MAX)
+                v->sentinel = mfree(v->sentinel);
+        else
+                v->sentinel = NULL;
 }
 
 static void varlink_clear(sd_varlink *v) {
@@ -1383,6 +1389,19 @@ static int varlink_format_queue(sd_varlink *v) {
         return 0;
 }
 
+static int varlink_enqueue_item(sd_varlink *v, VarlinkJsonQueueItem *q) {
+        assert(v);
+        assert(q);
+
+        if (v->n_output_queue >= VARLINK_QUEUE_MAX)
+                return -ENOBUFS;
+
+        LIST_INSERT_AFTER(queue, v->output_queue, v->output_queue_tail, q);
+        v->output_queue_tail = q;
+        v->n_output_queue++;
+        return 0;
+}
+
 static int varlink_enqueue_json(sd_varlink *v, sd_json_variant *m) {
         VarlinkJsonQueueItem *q;
 
@@ -1404,9 +1423,9 @@ static int varlink_enqueue_json(sd_varlink *v, sd_json_variant *m) {
 
         v->n_pushed_fds = 0; /* fds now belong to the queue entry */
 
-        LIST_INSERT_AFTER(queue, v->output_queue, v->output_queue_tail, q);
-        v->output_queue_tail = q;
-        v->n_output_queue++;
+        /* We already checked the precondition ourselves so this call cannot fail. */
+        assert_se(varlink_enqueue_item(v, q) >= 0);
+
         return 0;
 }
 
@@ -1536,16 +1555,46 @@ static int varlink_dispatch_method(sd_varlink *v) {
 
                 if (!invalid) {
                         r = callback(v, parameters, flags, v->userdata);
-                        if (r < 0 && VARLINK_STATE_WANTS_REPLY(v->state)) {
-                                varlink_log_errno(v, r, "Callback for %s returned error: %m", method);
+                        if (VARLINK_STATE_WANTS_REPLY(v->state)) {
+                                if (r < 0) {
+                                        varlink_log_errno(v, r, "Callback for %s returned error: %m", method);
 
-                                /* We got an error back from the callback. Propagate it to the client if the
-                                 * method call remains unanswered. */
-                                r = sd_varlink_error_errno(v, r);
-                                /* If we didn't manage to enqueue an error response, then fail the connection completely. */
+                                        /* We got an error back from the callback. Propagate it to the client
+                                         * if the method call remains unanswered. */
+                                        r = sd_varlink_error_errno(v, r);
+                                } else if (v->sentinel) {
+                                        if (v->previous) {
+                                                r = varlink_enqueue_item(v, v->previous);
+                                                if (r >= 0) {
+                                                        TAKE_PTR(v->previous);
+                                                        varlink_set_state(v, VARLINK_PROCESSED_METHOD);
+                                                }
+                                        } else {
+                                                char *sentinel = TAKE_PTR(v->sentinel);
+
+                                                /* Propagate the sentinel to the client if one was configured
+                                                 * and no replies were enqueued by the callback. */
+                                                if (sentinel == POINTER_MAX)
+                                                        r = sd_varlink_reply(v, NULL);
+                                                else
+                                                        r = sd_varlink_error(v, sentinel, NULL);
+
+                                                if (sentinel != POINTER_MAX)
+                                                        free(sentinel);
+                                        }
+                                        if (r < 0)
+                                                varlink_log_errno(v, r, "Failed to process sentinel for method '%s': %m", method);
+                                } else {
+                                        assert(!v->previous);
+                                        r = 0;
+                                }
+
+                                /* If we didn't manage to enqueue a response, then fail the connection completely. */
                                 if (r < 0 && VARLINK_STATE_WANTS_REPLY(v->state))
                                         goto fail;
-                        }
+
+                        } else
+                                assert(!v->previous);
                 }
         } else if (VARLINK_STATE_WANTS_REPLY(v->state)) {
                 r = sd_varlink_errorbo(v, SD_VARLINK_ERROR_METHOD_NOT_FOUND, SD_JSON_BUILD_PAIR_STRING("method", method));
@@ -2501,33 +2550,58 @@ _public_ int sd_varlink_collectb(
 }
 
 _public_ int sd_varlink_reply(sd_varlink *v, sd_json_variant *parameters) {
-        _cleanup_(sd_json_variant_unrefp) sd_json_variant *m = NULL;
         int r;
 
         assert_return(v, -EINVAL);
 
         if (v->state == VARLINK_DISCONNECTED)
-                return -ENOTCONN;
+                return varlink_log_errno(v, SYNTHETIC_ERRNO(ENOTCONN), "Not connected.");
+
         if (!IN_SET(v->state,
                     VARLINK_PROCESSING_METHOD, VARLINK_PROCESSING_METHOD_MORE,
                     VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE))
-                return -EBUSY;
+                return varlink_log_errno(v, SYNTHETIC_ERRNO(EBUSY), "Connection busy.");
+
+        bool more = IN_SET(v->state, VARLINK_PROCESSING_METHOD_MORE, VARLINK_PENDING_METHOD_MORE);
 
         /* Validate parameters BEFORE sanitization */
         if (v->current_method) {
                 const char *bad_field = NULL;
 
-                r = varlink_idl_validate_method_reply(v->current_method, parameters, /* flags= */ 0, &bad_field);
-                if (r < 0)
+                r = varlink_idl_validate_method_reply(v->current_method, parameters, more && v->sentinel ? SD_VARLINK_REPLY_CONTINUES : 0, &bad_field);
+                if (r == -EBADE)
+                        varlink_log_errno(v, r, "Method reply for %s() has 'continues' flag set, but IDL structure doesn't allow that, ignoring: %m",
+                                          v->current_method->name);
+                else if (r < 0)
                         /* Please adjust test/units/end.sh when updating the log message. */
                         varlink_log_errno(v, r, "Return parameters for method reply %s() didn't pass validation on field '%s', ignoring: %m",
                                           v->current_method->name, strna(bad_field));
         }
 
+        _cleanup_(sd_json_variant_unrefp) sd_json_variant *m = NULL;
         r = sd_json_buildo(&m, JSON_BUILD_PAIR_VARIANT_NON_EMPTY("parameters", parameters));
         if (r < 0)
                 return varlink_log_errno(v, r, "Failed to build json message: %m");
 
+        if (more && v->sentinel) {
+                if (v->previous) {
+                        r = sd_json_variant_set_field_boolean(&v->previous->data, "continues", true);
+                        if (r < 0)
+                                return r;
+
+                        r = varlink_enqueue_item(v, v->previous);
+                        if (r < 0)
+                                return varlink_log_errno(v, r, "Failed to enqueue json message: %m");
+                }
+
+                v->previous = varlink_json_queue_item_new(m, v->pushed_fds, v->n_pushed_fds);
+                if (!v->previous)
+                        return -ENOMEM;
+
+                v->n_pushed_fds = 0; /* fds now belong to the queue entry */
+                return 1;
+        }
+
         r = varlink_enqueue_json(v, m);
         if (r < 0)
                 return varlink_log_errno(v, r, "Failed to enqueue json message: %m");
@@ -2590,6 +2664,21 @@ _public_ int sd_varlink_error(sd_varlink *v, const char *error_id, sd_json_varia
                     VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE))
                 return varlink_log_errno(v, SYNTHETIC_ERRNO(EBUSY), "Connection busy.");
 
+        if (v->previous) {
+                r = sd_json_variant_set_field_boolean(&v->previous->data, "continues", true);
+                if (r < 0)
+                        return r;
+
+                /* If we have a previous reply still ready make sure we queue it before the error. We only
+                 * ever set "previous" if we're in a streaming method so we pass more=true uncondtionally
+                 * here as we know we're still going to queue an error afterwards. */
+                r = varlink_enqueue_item(v, v->previous);
+                if (r < 0)
+                        return varlink_log_errno(v, r, "Failed to enqueue json message: %m");
+
+                TAKE_PTR(v->previous);
+        }
+
         /* Reset the list of pushed file descriptors before sending an error reply. We do this here to
          * simplify code that puts together a complex reply message with fds, and half-way something
          * fails. In that case the pushed fds need to be flushed out again. Under the assumption that it
@@ -2721,6 +2810,11 @@ _public_ int sd_varlink_notify(sd_varlink *v, sd_json_variant *parameters) {
 
         assert_return(v, -EINVAL);
 
+        if (v->sentinel)
+                return varlink_log_errno(v, SYNTHETIC_ERRNO(EINVAL), "Cannot use sd_varlink_notify() on method with sentinel set");
+
+        assert(!v->previous);
+
         if (v->state == VARLINK_DISCONNECTED)
                 return varlink_log_errno(v, SYNTHETIC_ERRNO(ENOTCONN), "Not connected.");
 

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

@@ -82,7 +82,7 @@ struct VarlinkJsonQueueItem {
         int fds[];
 };
 
-struct sd_varlink {
+typedef struct sd_varlink {
         unsigned n_ref;
 
         sd_varlink_server *server;
@@ -157,6 +157,9 @@ struct sd_varlink {
         sd_varlink_reply_flags_t current_reply_flags;
         sd_varlink_symbol *current_method;
 
+        VarlinkJsonQueueItem *previous;
+        char *sentinel;
+
         int peer_pidfd;
         struct ucred ucred;
         bool ucred_acquired:1;
@@ -189,7 +192,7 @@ struct sd_varlink {
         sd_event_source *defer_event_source;
 
         PidRef exec_pidref;
-};
+} sd_varlink;
 
 typedef struct VarlinkServerSocket VarlinkServerSocket;
 
@@ -204,7 +207,7 @@ struct VarlinkServerSocket {
         LIST_FIELDS(VarlinkServerSocket, sockets);
 };
 
-struct sd_varlink_server {
+typedef struct sd_varlink_server {
         unsigned n_ref;
         sd_varlink_server_flags_t flags;
 
@@ -234,7 +237,7 @@ struct sd_varlink_server {
         unsigned connections_per_uid_max;
 
         bool exit_on_idle;
-};
+} sd_varlink_server;
 
 #define varlink_log_errno(v, error, fmt, ...)                           \
         log_debug_errno(error, "%s: " fmt, varlink_description(v), ##__VA_ARGS__)

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

@@ -6,6 +6,7 @@
 #include "pidref.h"
 #include "set.h"
 #include "string-util.h"
+#include "varlink-internal.h"
 #include "varlink-util.h"
 #include "version.h"
 
@@ -204,6 +205,31 @@ int varlink_check_privileged_peer(sd_varlink *vl) {
         return 0;
 }
 
+int varlink_set_sentinel(sd_varlink *v, const char *error_id) {
+        _cleanup_free_ char *s = NULL;
+
+        assert(v);
+
+        /* If the caller doesn't want a reply, then don't set a sentinel. */
+        if (v->state == VARLINK_PROCESSING_METHOD_ONEWAY)
+                return 0;
+
+        /* This has to be called during a callback, and not after it has exited. */
+        assert(IN_SET(v->state, VARLINK_PROCESSING_METHOD, VARLINK_PROCESSING_METHOD_MORE));
+
+        if (error_id) {
+                s = strdup(error_id);
+                if (!s)
+                        return -ENOMEM;
+        }
+
+        if (v->sentinel != POINTER_MAX)
+                free(v->sentinel);
+
+        v->sentinel = s ? TAKE_PTR(s) : POINTER_MAX;
+        return 0;
+}
+
 DEFINE_HASH_OPS_WITH_VALUE_DESTRUCTOR(
                 varlink_hash_ops,
                 void,

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

@@ -28,4 +28,6 @@ int varlink_server_new(
 
 int varlink_check_privileged_peer(sd_varlink *vl);
 
+int varlink_set_sentinel(sd_varlink *v, const char *error_id);
+
 extern const struct hash_ops varlink_hash_ops;