sd-varlink: Allow using sd_varlink_reply() in streaming methods (#40546)

9e10f3a7e800ad67be8d8b14ae158a27438814f0 changed the implementation
to report an error instead of an empty object but the IDL was not
adjusted. Let's fix that.

src/bootctl/bootctl-status.c

@@ -22,6 +22,7 @@
 #include "pretty-print.h"
 #include "string-util.h"
 #include "tpm2-util.h"
+#include "varlink-util.h"
 
 static int status_entries(
                 const BootConfig *config,
@@ -674,23 +675,21 @@ int vl_method_list_boot_entries(sd_varlink *link, sd_json_variant *parameters, s
         if (r < 0)
                 return r;
 
-        _cleanup_(sd_json_variant_unrefp) sd_json_variant *previous = NULL;
+        r = varlink_set_sentinel(link, "io.systemd.BootControl.NoSuchBootEntry");
+        if (r < 0)
+                return r;
+
         for (size_t i = 0; i < config.n_entries; i++) {
-                if (previous) {
-                        r = sd_varlink_notifybo(link, SD_JSON_BUILD_PAIR_VARIANT("entry", previous));
-                        if (r < 0)
-                                return r;
+                _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL;
 
-                        previous = sd_json_variant_unref(previous);
-                }
+                r = boot_entry_to_json(&config, i, &v);
+                if (r < 0)
+                        return r;
 
-                r = boot_entry_to_json(&config, i, &previous);
+                r = sd_varlink_replybo(link, SD_JSON_BUILD_PAIR_VARIANT("entry", v));
                 if (r < 0)
                         return r;
         }
 
-        if (!previous)
-                return sd_varlink_error(link, "io.systemd.BootControl.NoSuchBootEntry", NULL);
-
-        return sd_varlink_replybo(link, SD_JSON_BUILD_PAIR_VARIANT("entry", previous));
+        return 0;
 }

src/core/varlink-dynamic-user.c

@@ -10,6 +10,7 @@
 #include "uid-classification.h"
 #include "user-util.h"
 #include "varlink-dynamic-user.h"
+#include "varlink-util.h"
 
 typedef struct LookupParameters {
         const char *user_name;
@@ -59,7 +60,6 @@ int vl_method_get_user_record(sd_varlink *link, sd_json_variant *parameters, sd_
                 {}
         };
 
-        _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL;
         LookupParameters p = {
                 .uid = UID_INVALID,
         };
@@ -78,6 +78,10 @@ int vl_method_get_user_record(sd_varlink *link, sd_json_variant *parameters, sd_
         if (!streq_ptr(p.service, "io.systemd.DynamicUser"))
                 return sd_varlink_error(link, "io.systemd.UserDatabase.BadService", NULL);
 
+        r = varlink_set_sentinel(link, "io.systemd.UserDatabase.NoRecordFound");
+        if (r < 0)
+                return r;
+
         if (uid_is_valid(p.uid))
                 r = dynamic_user_lookup_uid(m, p.uid, &found_name);
         else if (p.user_name)
@@ -98,26 +102,20 @@ int vl_method_get_user_record(sd_varlink *link, sd_json_variant *parameters, sd_
                         if (!user_match_lookup_parameters(&p, d->name, uid))
                                 continue;
 
-                        if (v) {
-                                r = sd_varlink_notify(link, v);
-                                if (r < 0)
-                                        return r;
-
-                                v = sd_json_variant_unref(v);
-                        }
-
+                        _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL;
                         r = build_user_json(d->name, uid, &v);
                         if (r < 0)
                                 return r;
+
+                        r = sd_varlink_reply(link, v);
+                        if (r < 0)
+                                return r;
                 }
 
-                if (!v)
-                        return sd_varlink_error(link, "io.systemd.UserDatabase.NoRecordFound", NULL);
-
-                return sd_varlink_reply(link, v);
+                return 0;
         }
         if (r == -ESRCH)
-                return sd_varlink_error(link, "io.systemd.UserDatabase.NoRecordFound", NULL);
+                return 0;
         if (r < 0)
                 return r;
 
@@ -127,6 +125,7 @@ int vl_method_get_user_record(sd_varlink *link, sd_json_variant *parameters, sd_
         if (!user_match_lookup_parameters(&p, un, uid))
                 return sd_varlink_error(link, "io.systemd.UserDatabase.ConflictingRecordFound", NULL);
 
+        _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL;
         r = build_user_json(un, uid, &v);
         if (r < 0)
                 return r;
@@ -168,7 +167,6 @@ int vl_method_get_group_record(sd_varlink *link, sd_json_variant *parameters, sd
                 {}
         };
 
-        _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL;
         LookupParameters p = {
                 .gid = GID_INVALID,
         };
@@ -184,6 +182,10 @@ int vl_method_get_group_record(sd_varlink *link, sd_json_variant *parameters, sd
         if (r != 0)
                 return r;
 
+        r = varlink_set_sentinel(link, "io.systemd.UserDatabase.NoRecordFound");
+        if (r < 0)
+                return r;
+
         if (!streq_ptr(p.service, "io.systemd.DynamicUser"))
                 return sd_varlink_error(link, "io.systemd.UserDatabase.BadService", NULL);
 
@@ -209,26 +211,20 @@ int vl_method_get_group_record(sd_varlink *link, sd_json_variant *parameters, sd
                         if (!group_match_lookup_parameters(&p, d->name, (gid_t) uid))
                                 continue;
 
-                        if (v) {
-                                r = sd_varlink_notify(link, v);
-                                if (r < 0)
-                                        return r;
-
-                                v = sd_json_variant_unref(v);
-                        }
-
+                        _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL;
                         r = build_group_json(d->name, (gid_t) uid, &v);
                         if (r < 0)
                                 return r;
+
+                        r = sd_varlink_reply(link, v);
+                        if (r < 0)
+                                return r;
                 }
 
-                if (!v)
-                        return sd_varlink_error(link, "io.systemd.UserDatabase.NoRecordFound", NULL);
-
-                return sd_varlink_reply(link, v);
+                return 0;
         }
         if (r == -ESRCH)
-                return sd_varlink_error(link, "io.systemd.UserDatabase.NoRecordFound", NULL);
+                return 0;
         if (r < 0)
                 return r;
 
@@ -238,6 +234,7 @@ int vl_method_get_group_record(sd_varlink *link, sd_json_variant *parameters, sd
         if (!group_match_lookup_parameters(&p, gn, gid))
                 return sd_varlink_error(link, "io.systemd.UserDatabase.ConflictingRecordFound", NULL);
 
+        _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL;
         r = build_group_json(gn, gid, &v);
         if (r < 0)
                 return r;

src/core/varlink-manager.c

@@ -330,6 +330,10 @@ int vl_method_enqueue_marked_jobs_manager(sd_varlink *link, sd_json_variant *par
         if (r <= 0)
                 return r;
 
+        r = varlink_set_sentinel(link, NULL);
+        if (r < 0)
+                return r;
+
         log_info("Queuing reload/restart jobs for marked units%s", glyph(GLYPH_ELLIPSIS));
 
         Unit *u;
@@ -373,20 +377,17 @@ int vl_method_enqueue_marked_jobs_manager(sd_varlink *link, sd_json_variant *par
 
                         const char *error_msg = bus_error.message ?: error_id ? NULL : STRERROR(r);
 
-                        r = sd_varlink_notifybo(link,
-                                                SD_JSON_BUILD_PAIR_STRING("unitID", u->id),
-                                                JSON_BUILD_PAIR_STRING_NON_EMPTY("error", error_id),
-                                                JSON_BUILD_PAIR_STRING_NON_EMPTY("errorMessage", error_msg));
+                        r = sd_varlink_replybo(link,
+                                               SD_JSON_BUILD_PAIR_STRING("unitID", u->id),
+                                               JSON_BUILD_PAIR_STRING_NON_EMPTY("error", error_id),
+                                               JSON_BUILD_PAIR_STRING_NON_EMPTY("errorMessage", error_msg));
                 } else
-                        r = sd_varlink_notifybo(link,
-                                                SD_JSON_BUILD_PAIR_STRING("unitID", u->id),
-                                                SD_JSON_BUILD_PAIR_INTEGER("jobID", job_id));
+                        r = sd_varlink_replybo(link,
+                                               SD_JSON_BUILD_PAIR_STRING("unitID", u->id),
+                                               SD_JSON_BUILD_PAIR_INTEGER("jobID", job_id));
                 if (r < 0)
                         return r;
         }
 
-        if (ret < 0)
-                return ret;
-
-        return sd_varlink_reply(link, NULL);
+        return ret;
 }

src/core/varlink-metrics.c

@@ -25,7 +25,7 @@ static int unit_active_state_build_json(MetricFamilyContext *context, void *user
                                 context,
                                 unit->id,
                                 unit_active_state_to_string(unit_active_state(unit)),
-                                /* field_pairs= */ NULL);
+                                /* fields= */ NULL);
                 if (r < 0)
                         return r;
         }
@@ -50,7 +50,7 @@ static int unit_load_state_build_json(MetricFamilyContext *context, void *userda
                                 context,
                                 unit->id,
                                 unit_load_state_to_string(unit->load_state),
-                                /* field_pairs= */ NULL);
+                                /* fields= */ NULL);
                 if (r < 0)
                         return r;
         }
@@ -66,7 +66,7 @@ static int nrestarts_build_json(MetricFamilyContext *context, void *userdata) {
 
         LIST_FOREACH(units_by_type, unit, manager->units_by_type[UNIT_SERVICE]) {
                 r = metric_build_send_unsigned(
-                                context, unit->id, SERVICE(unit)->n_restarts, /* field_pairs= */ NULL);
+                                context, unit->id, SERVICE(unit)->n_restarts, /* fields= */ NULL);
                 if (r < 0)
                         return r;
         }
@@ -81,16 +81,21 @@ static int units_by_type_total_build_json(MetricFamilyContext *context, void *us
         assert(context);
 
         for (UnitType type = 0; type < _UNIT_TYPE_MAX; type++) {
+                _cleanup_(sd_json_variant_unrefp) sd_json_variant *fields = NULL;
                 uint64_t counter = 0;
 
                 LIST_FOREACH(units_by_type, _u, manager->units_by_type[type])
                         counter++;
 
+                r = sd_json_buildo(&fields, SD_JSON_BUILD_PAIR_STRING("type", unit_type_to_string(type)));
+                if (r < 0)
+                        return r;
+
                 r = metric_build_send_unsigned(
                                 context,
                                 /* object= */ NULL,
                                 counter,
-                                STRV_MAKE("type", unit_type_to_string(type)));
+                                fields);
                 if (r < 0)
                         return r;
         }
@@ -117,11 +122,17 @@ static int units_by_state_total_build_json(MetricFamilyContext *context, void *u
         }
 
         for (UnitActiveState state = 0; state < _UNIT_ACTIVE_STATE_MAX; state++) {
+                _cleanup_(sd_json_variant_unrefp) sd_json_variant *fields = NULL;
+
+                r = sd_json_buildo(&fields, SD_JSON_BUILD_PAIR_STRING("state", unit_active_state_to_string(state)));
+                if (r < 0)
+                        return r;
+
                 r = metric_build_send_unsigned(
                                 context,
                                 /* object= */ NULL,
                                 counters[state],
-                                STRV_MAKE("state", unit_active_state_to_string(state)));
+                                fields);
                 if (r < 0)
                         return r;
         }
@@ -135,31 +146,31 @@ const MetricFamily metric_family_table[] = {
          .name = METRIC_IO_SYSTEMD_MANAGER_PREFIX "nrestarts",
          .description = "Per unit metric: number of restarts",
          .type = METRIC_FAMILY_TYPE_COUNTER,
-         .generate_cb = nrestarts_build_json,
+         .generate = nrestarts_build_json,
         },
         {
          .name = METRIC_IO_SYSTEMD_MANAGER_PREFIX "unit_active_state",
          .description = "Per unit metric: active state",
          .type = METRIC_FAMILY_TYPE_STRING,
-         .generate_cb = unit_active_state_build_json,
+         .generate = unit_active_state_build_json,
         },
         {
          .name = METRIC_IO_SYSTEMD_MANAGER_PREFIX "unit_load_state",
          .description = "Per unit metric: load state",
          .type = METRIC_FAMILY_TYPE_STRING,
-         .generate_cb = unit_load_state_build_json,
+         .generate = unit_load_state_build_json,
         },
         {
          .name = METRIC_IO_SYSTEMD_MANAGER_PREFIX "units_by_state_total",
          .description = "Total number of units of different state",
          .type = METRIC_FAMILY_TYPE_GAUGE,
-         .generate_cb = units_by_state_total_build_json,
+         .generate = units_by_state_total_build_json,
         },
         {
          .name = METRIC_IO_SYSTEMD_MANAGER_PREFIX "units_by_type_total",
          .description = "Total number of units of different types",
          .type = METRIC_FAMILY_TYPE_GAUGE,
-         .generate_cb = units_by_type_total_build_json,
+         .generate = units_by_type_total_build_json,
         },
         {}
 };

src/core/varlink-unit.c

@@ -311,27 +311,16 @@ static int unit_runtime_build_json(sd_json_variant **ret, const char *name, void
                         JSON_BUILD_PAIR_CALLBACK_NON_NULL("CGroup", unit_cgroup_runtime_build_json, u));
 }
 
-static int list_unit_one(sd_varlink *link, Unit *unit, bool more) {
-        _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL;
-        int r;
-
+static int list_unit_one(sd_varlink *link, Unit *unit) {
         assert(link);
         assert(unit);
 
-        r = sd_json_buildo(
-                &v,
-                SD_JSON_BUILD_PAIR_CALLBACK("context", unit_context_build_json, unit),
-                SD_JSON_BUILD_PAIR_CALLBACK("runtime", unit_runtime_build_json, unit));
-        if (r < 0)
-                return r;
-
-        if (more)
-                return sd_varlink_notify(link, v);
-
-        return sd_varlink_reply(link, v);
+        return sd_varlink_replybo(link,
+                        SD_JSON_BUILD_PAIR_CALLBACK("context", unit_context_build_json, unit),
+                        SD_JSON_BUILD_PAIR_CALLBACK("runtime", unit_runtime_build_json, unit));
 }
 
-static int list_unit_one_with_selinux_access_check(sd_varlink *link, Unit *unit, bool more) {
+static int list_unit_one_with_selinux_access_check(sd_varlink *link, Unit *unit) {
         int r;
 
         assert(link);
@@ -343,7 +332,7 @@ static int list_unit_one_with_selinux_access_check(sd_varlink *link, Unit *unit,
                  * it means that SELinux enforce is on. It also does all the logging(). */
                 return sd_varlink_error(link, SD_VARLINK_ERROR_PERMISSION_DENIED, NULL);
 
-        return list_unit_one(link, unit, more);
+        return list_unit_one(link, unit);
 }
 
 static int lookup_unit_by_pidref(sd_varlink *link, Manager *manager, PidRef *pidref, Unit **ret_unit) {
@@ -395,7 +384,12 @@ static int varlink_error_conflict_lookup_parameters(sd_varlink *v, const UnitLoo
         return varlink_error_no_such_unit(v, /* name= */ NULL);
 }
 
-static int lookup_unit_by_parameters(sd_varlink *link, Manager *manager, UnitLookupParameters *p, Unit **ret_unit) {
+static int lookup_unit_by_parameters(
+                sd_varlink *link,
+                Manager *manager,
+                UnitLookupParameters *p,
+                Unit **ret) {
+
         /* The function can return ret_unit=NULL if no lookup parameters provided */
         Unit *unit = NULL;
         int r;
@@ -403,7 +397,7 @@ static int lookup_unit_by_parameters(sd_varlink *link, Manager *manager, UnitLoo
         assert(link);
         assert(manager);
         assert(p);
-        assert(ret_unit);
+        assert(ret);
 
         if (p->name) {
                 unit = manager_get_unit(manager, p->name);
@@ -413,6 +407,7 @@ static int lookup_unit_by_parameters(sd_varlink *link, Manager *manager, UnitLoo
 
         if (pidref_is_set_or_automatic(&p->pidref)) {
                 Unit *pid_unit;
+
                 r = lookup_unit_by_pidref(link, manager, &p->pidref, &pid_unit);
                 if (r == -EINVAL)
                         return sd_varlink_error_invalid_parameter_name(link, "pid");
@@ -420,7 +415,7 @@ static int lookup_unit_by_parameters(sd_varlink *link, Manager *manager, UnitLoo
                         return varlink_error_no_such_unit(link, "pid");
                 if (r < 0)
                         return r;
-                if (pid_unit != unit && unit != NULL)
+                if (unit && pid_unit != unit)
                         return varlink_error_conflict_lookup_parameters(link, p);
 
                 unit = pid_unit;
@@ -433,7 +428,7 @@ static int lookup_unit_by_parameters(sd_varlink *link, Manager *manager, UnitLoo
                 Unit *cgroup_unit = manager_get_unit_by_cgroup(manager, p->cgroup);
                 if (!cgroup_unit)
                         return varlink_error_no_such_unit(link, "cgroup");
-                if (cgroup_unit != unit && unit != NULL)
+                if (unit && cgroup_unit != unit)
                         return varlink_error_conflict_lookup_parameters(link, p);
 
                 unit = cgroup_unit;
@@ -443,14 +438,14 @@ static int lookup_unit_by_parameters(sd_varlink *link, Manager *manager, UnitLoo
                 Unit *id128_unit = hashmap_get(manager->units_by_invocation_id, &p->invocation_id);
                 if (!id128_unit)
                         return varlink_error_no_such_unit(link, "invocationID");
-                if (id128_unit != unit && unit != NULL)
+                if (unit && id128_unit != unit)
                         return varlink_error_conflict_lookup_parameters(link, p);
 
                 unit = id128_unit;
         }
 
-        *ret_unit = unit;
-        return 0;
+        *ret = unit;
+        return !!unit;
 }
 
 int vl_method_list_units(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata) {
@@ -466,7 +461,7 @@ int vl_method_list_units(sd_varlink *link, sd_json_variant *parameters, sd_varli
          _cleanup_(unit_lookup_parameters_done) UnitLookupParameters p = {
                  .pidref = PIDREF_NULL,
         };
-        Unit *unit, *previous = NULL;
+        Unit *unit;
         const char *k;
         int r;
 
@@ -480,30 +475,27 @@ int vl_method_list_units(sd_varlink *link, sd_json_variant *parameters, sd_varli
         r = lookup_unit_by_parameters(link, manager, &p, &unit);
         if (r < 0)
                 return r;
-        if (unit)
-                return list_unit_one_with_selinux_access_check(link, unit, /* more= */ false);
+        if (r > 0)
+                return list_unit_one_with_selinux_access_check(link, unit);
 
         if (!FLAGS_SET(flags, SD_VARLINK_METHOD_MORE))
                 return sd_varlink_error(link, SD_VARLINK_ERROR_EXPECTED_MORE, NULL);
 
+        r = varlink_set_sentinel(link, "io.systemd.Manager.NoSuchUnit");
+        if (r < 0)
+                return r;
+
         HASHMAP_FOREACH_KEY(unit, k, manager->units) {
                 /* ignore aliases */
                 if (k != unit->id)
                         continue;
 
-                if (previous) {
-                        r = list_unit_one(link, previous, /* more= */ true);
-                        if (r < 0)
-                                return r;
-                }
-
-                previous = unit;
+                r = list_unit_one(link, unit);
+                if (r < 0)
+                        return r;
         }
 
-        if (previous)
-                return list_unit_one(link, previous, /* more= */ false);
-
-        return sd_varlink_error(link, "io.systemd.Manager.NoSuchUnit", NULL);
+        return 0;
 }
 
 int varlink_unit_queue_job_one(

src/home/homed-varlink.c

@@ -14,6 +14,7 @@
 #include "user-record.h"
 #include "user-record-util.h"
 #include "user-util.h"
+#include "varlink-util.h"
 
 typedef struct LookupParameters {
         const char *user_name;
@@ -86,7 +87,6 @@ int vl_method_get_user_record(sd_varlink *link, sd_json_variant *parameters, sd_
                 {}
         };
 
-        _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL;
         LookupParameters p = {
                 .uid = UID_INVALID,
         };
@@ -104,6 +104,10 @@ int vl_method_get_user_record(sd_varlink *link, sd_json_variant *parameters, sd_
         if (!streq_ptr(p.service, m->userdb_service))
                 return sd_varlink_error(link, "io.systemd.UserDatabase.BadService", NULL);
 
+        r = varlink_set_sentinel(link, "io.systemd.UserDatabase.NoRecordFound");
+        if (r < 0)
+                return r;
+
         if (uid_is_valid(p.uid))
                 h = hashmap_get(m->homes_by_uid, UID_TO_PTR(p.uid));
         else if (p.user_name) {
@@ -112,45 +116,36 @@ int vl_method_get_user_record(sd_varlink *link, sd_json_variant *parameters, sd_
                         return r;
         } else {
 
-                /* If neither UID nor name was specified, then dump all homes. Do so with varlink_notify()
-                 * for all entries but the last, so that clients can stream the results, and easily process
-                 * them piecemeal. */
+                /* If neither UID nor name was specified, then dump all homes. */
 
                 HASHMAP_FOREACH(h, m->homes_by_uid) {
-
                         if (!home_user_match_lookup_parameters(&p, h))
                                 continue;
 
-                        if (v) {
-                                /* An entry set from the previous iteration? Then send it now */
-                                r = sd_varlink_notify(link, v);
-                                if (r < 0)
-                                        return r;
-
-                                v = sd_json_variant_unref(v);
-                        }
-
                         trusted = client_is_trusted(link, h);
 
+                        _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL;
                         r = build_user_json(h, trusted, &v);
                         if (r < 0)
                                 return r;
+
+                        r = sd_varlink_reply(link, v);
+                        if (r < 0)
+                                return r;
                 }
 
-                if (!v)
-                        return sd_varlink_error(link, "io.systemd.UserDatabase.NoRecordFound", NULL);
-
-                return sd_varlink_reply(link, v);
+                return 0;
         }
 
         if (!h)
-                return sd_varlink_error(link, "io.systemd.UserDatabase.NoRecordFound", NULL);
+                return 0;
 
         if (!home_user_match_lookup_parameters(&p, h))
                 return sd_varlink_error(link, "io.systemd.UserDatabase.ConflictingRecordFound", NULL);
 
         trusted = client_is_trusted(link, h);
 
+        _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL;
         r = build_user_json(h, trusted, &v);
         if (r < 0)
                 return r;
@@ -201,7 +196,6 @@ int vl_method_get_group_record(sd_varlink *link, sd_json_variant *parameters, sd
                 {}
         };
 
-        _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL;
         LookupParameters p = {
                 .gid = GID_INVALID,
         };
@@ -218,6 +212,10 @@ int vl_method_get_group_record(sd_varlink *link, sd_json_variant *parameters, sd
         if (!streq_ptr(p.service, m->userdb_service))
                 return sd_varlink_error(link, "io.systemd.UserDatabase.BadService", NULL);
 
+        r = varlink_set_sentinel(link, "io.systemd.UserDatabase.NoRecordFound");
+        if (r < 0)
+                return r;
+
         if (gid_is_valid(p.gid))
                 h = hashmap_get(m->homes_by_uid, UID_TO_PTR((uid_t) p.gid));
         else if (p.group_name) {
@@ -225,37 +223,30 @@ int vl_method_get_group_record(sd_varlink *link, sd_json_variant *parameters, sd
                 if (r < 0)
                         return r;
         } else {
-
                 HASHMAP_FOREACH(h, m->homes_by_uid) {
-
                         if (!home_group_match_lookup_parameters(&p, h))
                                 continue;
 
-                        if (v) {
-                                r = sd_varlink_notify(link, v);
-                                if (r < 0)
-                                        return r;
-
-                                v = sd_json_variant_unref(v);
-                        }
-
+                        _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL;
                         r = build_group_json(h, &v);
                         if (r < 0)
                                 return r;
+
+                        r = sd_varlink_reply(link, v);
+                        if (r < 0)
+                                return r;
                 }
 
-                if (!v)
-                        return sd_varlink_error(link, "io.systemd.UserDatabase.NoRecordFound", NULL);
-
-                return sd_varlink_reply(link, v);
+                return 0;
         }
 
         if (!h)
-                return sd_varlink_error(link, "io.systemd.UserDatabase.NoRecordFound", NULL);
+                return 0;
 
         if (!home_group_match_lookup_parameters(&p, h))
                 return sd_varlink_error(link, "io.systemd.UserDatabase.ConflictingRecordFound", NULL);
 
+        _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL;
         r = build_group_json(h, &v);
         if (r < 0)
                 return r;
@@ -286,17 +277,21 @@ int vl_method_get_memberships(sd_varlink *link, sd_json_variant *parameters, sd_
         if (!streq_ptr(p.service, m->userdb_service))
                 return sd_varlink_error(link, "io.systemd.UserDatabase.BadService", NULL);
 
+        r = varlink_set_sentinel(link, "io.systemd.UserDatabase.NoRecordFound");
+        if (r < 0)
+                return r;
+
         if (p.user_name) {
                 r = manager_get_home_by_name(m, p.user_name, &h);
                 if (r < 0)
                         return r;
                 if (!h)
-                        return sd_varlink_error(link, "io.systemd.UserDatabase.NoRecordFound", NULL);
+                        return 0;
 
                 if (p.group_name) {
                         if (!strv_contains(h->record->member_of, p.group_name) &&
                             !user_record_matches_user_name(h->record, p.group_name))
-                                return sd_varlink_error(link, "io.systemd.UserDatabase.NoRecordFound", NULL);
+                                return 0;
 
                         return sd_varlink_replybo(
                                         link,
@@ -305,7 +300,7 @@ int vl_method_get_memberships(sd_varlink *link, sd_json_variant *parameters, sd_
                 }
 
                 STRV_FOREACH(i, h->record->member_of) {
-                        r = sd_varlink_notifybo(
+                        r = sd_varlink_replybo(
                                         link,
                                         SD_JSON_BUILD_PAIR_STRING("userName", h->user_name),
                                         SD_JSON_BUILD_PAIR_STRING("groupName", *i));
@@ -319,64 +314,37 @@ int vl_method_get_memberships(sd_varlink *link, sd_json_variant *parameters, sd_
                                 SD_JSON_BUILD_PAIR_STRING("groupName", h->user_name));
 
         } else if (p.group_name) {
-                const char *last = NULL;
-
                 HASHMAP_FOREACH(h, m->homes_by_uid) {
-
                         if (!strv_contains(h->record->member_of, p.group_name) &&
                             !user_record_matches_user_name(h->record, p.group_name))
                                 continue;
 
-                        if (last) {
-                                r = sd_varlink_notifybo(
-                                                link,
-                                                SD_JSON_BUILD_PAIR_STRING("userName", last),
-                                                SD_JSON_BUILD_PAIR_STRING("groupName", p.group_name));
-                                if (r < 0)
-                                        return r;
-                        }
-
-                        last = h->user_name;
-                }
-
-                if (last)
-                        return sd_varlink_replybo(
+                        r = sd_varlink_replybo(
                                         link,
-                                        SD_JSON_BUILD_PAIR_STRING("userName", last),
+                                        SD_JSON_BUILD_PAIR_STRING("userName", h->user_name),
                                         SD_JSON_BUILD_PAIR_STRING("groupName", p.group_name));
+                        if (r < 0)
+                                return r;
+                }
         } else {
-                const char *last = NULL;
-
                 HASHMAP_FOREACH(h, m->homes_by_uid) {
+                        r = sd_varlink_replybo(
+                                        link,
+                                        SD_JSON_BUILD_PAIR_STRING("userName", h->user_name),
+                                        SD_JSON_BUILD_PAIR_STRING("groupName", h->user_name));
+                        if (r < 0)
+                                return r;
+
                         STRV_FOREACH(j, h->record->member_of) {
-                                if (last) {
-                                        r = sd_varlink_notifybo(
-                                                        link,
-                                                        SD_JSON_BUILD_PAIR_STRING("userName", last),
-                                                        SD_JSON_BUILD_PAIR_STRING("groupName", last));
-                                        if (r < 0)
-                                                return r;
-
-                                        last = NULL;
-                                }
-
-                                r = sd_varlink_notifybo(
+                                r = sd_varlink_replybo(
                                                 link,
                                                 SD_JSON_BUILD_PAIR_STRING("userName", h->user_name),
                                                 SD_JSON_BUILD_PAIR_STRING("groupName", *j));
                                 if (r < 0)
                                         return r;
                         }
-
-                        last = h->user_name;
                 }
-
-                if (last)
-                        return sd_varlink_replybo(
-                                        link,
-                                        SD_JSON_BUILD_PAIR_STRING("userName", last),
-                                        SD_JSON_BUILD_PAIR_STRING("groupName", last));
         }
 
-        return sd_varlink_error(link, "io.systemd.UserDatabase.NoRecordFound", NULL);
+        return 0;
 }

src/import/importd.c

@@ -1801,38 +1801,26 @@ static int vl_method_list_transfers(sd_varlink *link, sd_json_variant *parameter
         if (!FLAGS_SET(flags, SD_VARLINK_METHOD_MORE))
                 return sd_varlink_error(link, SD_VARLINK_ERROR_EXPECTED_MORE, NULL);
 
-        Transfer *previous = NULL, *t;
-        HASHMAP_FOREACH(t, m->transfers) {
+        r = varlink_set_sentinel(link, "io.systemd.Import.NoTransfers");
+        if (r < 0)
+                return r;
 
+        Transfer *t;
+        HASHMAP_FOREACH(t, m->transfers) {
                 if (p.class >= 0 && p.class != t->class)
                         continue;
 
-                if (previous) {
-                        _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL;
-
-                        r = make_transfer_json(previous, &v);
-                        if (r < 0)
-                                return r;
-
-                        r = sd_varlink_notify(link, v);
-                        if (r < 0)
-                                return r;
-                }
-
-                previous = t;
-        }
-
-        if (previous) {
                 _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL;
-
-                r = make_transfer_json(previous, &v);
+                r = make_transfer_json(t, &v);
                 if (r < 0)
                         return r;
 
-                return sd_varlink_reply(link, v);
+                r = sd_varlink_reply(link, v);
+                if (r < 0)
+                        return r;
         }
 
-        return sd_varlink_error(link, "io.systemd.Import.NoTransfers", NULL);
+        return 0;
 }
 
 static JSON_DISPATCH_ENUM_DEFINE(json_dispatch_import_verify, ImportVerify, import_verify_from_string);

src/libsystemd/sd-common/sd-forward.h

@@ -119,6 +119,8 @@ typedef struct sd_varlink_field sd_varlink_field;
 typedef struct sd_varlink_symbol sd_varlink_symbol;
 typedef struct sd_varlink_interface sd_varlink_interface;
 
+typedef int (*sd_varlink_method_t)(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata);
+
 typedef struct sd_journal sd_journal;
 
 typedef struct sd_resolve sd_resolve;

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) {
@@ -1275,6 +1281,154 @@ static int generic_method_get_interface_description(
                         SD_JSON_BUILD_PAIR_STRING("description", text));
 }
 
+static int varlink_format_json(sd_varlink *v, sd_json_variant *m) {
+        _cleanup_(erase_and_freep) char *text = NULL;
+        int sz, r;
+
+        assert(v);
+        assert(m);
+
+        sz = sd_json_variant_format(m, /* flags= */ 0, &text);
+        if (sz < 0)
+                return sz;
+        assert(text[sz] == '\0');
+
+        if (v->output_buffer_size + sz + 1 > VARLINK_BUFFER_MAX)
+                return -ENOBUFS;
+
+        if (DEBUG_LOGGING) {
+                _cleanup_(erase_and_freep) char *censored_text = NULL;
+
+                /* Suppress sensitive fields in the debug output */
+                r = sd_json_variant_format(m, SD_JSON_FORMAT_CENSOR_SENSITIVE, &censored_text);
+                if (r < 0)
+                        return r;
+
+                varlink_log(v, "Sending message: %s", censored_text);
+        }
+
+        if (v->output_buffer_size == 0) {
+
+                free_and_replace(v->output_buffer, text);
+
+                v->output_buffer_size = sz + 1;
+                v->output_buffer_index = 0;
+
+        } else if (v->output_buffer_index == 0) {
+
+                if (!GREEDY_REALLOC(v->output_buffer, v->output_buffer_size + sz + 1))
+                        return -ENOMEM;
+
+                memcpy(v->output_buffer + v->output_buffer_size, text, sz + 1);
+                v->output_buffer_size += sz + 1;
+        } else {
+                char *n;
+                const size_t new_size = v->output_buffer_size + sz + 1;
+
+                n = new(char, new_size);
+                if (!n)
+                        return -ENOMEM;
+
+                memcpy(mempcpy(n, v->output_buffer + v->output_buffer_index, v->output_buffer_size), text, sz + 1);
+
+                free_and_replace(v->output_buffer, n);
+                v->output_buffer_size = new_size;
+                v->output_buffer_index = 0;
+        }
+
+        if (sd_json_variant_is_sensitive_recursive(m))
+                v->output_buffer_sensitive = true; /* Propagate sensitive flag */
+        else
+                text = mfree(text); /* No point in the erase_and_free() destructor declared above */
+
+        return 0;
+}
+
+static int varlink_format_queue(sd_varlink *v) {
+        int r;
+
+        assert(v);
+
+        /* Takes entries out of the output queue and formats them into the output buffer. But only if this
+         * would not corrupt our fd message boundaries */
+
+        while (v->output_queue) {
+                _cleanup_free_ int *array = NULL;
+
+                assert(v->n_output_queue > 0);
+
+                VarlinkJsonQueueItem *q = v->output_queue;
+
+                if (v->n_output_fds > 0) /* unwritten fds? if we'd add more we'd corrupt the fd message boundaries, hence wait */
+                        return 0;
+
+                if (q->n_fds > 0) {
+                        array = newdup(int, q->fds, q->n_fds);
+                        if (!array)
+                                return -ENOMEM;
+                }
+
+                r = varlink_format_json(v, q->data);
+                if (r < 0)
+                        return r;
+
+                /* Take possession of the queue element's fds */
+                free(v->output_fds);
+                v->output_fds = TAKE_PTR(array);
+                v->n_output_fds = q->n_fds;
+                q->n_fds = 0;
+
+                LIST_REMOVE(queue, v->output_queue, q);
+                if (!v->output_queue)
+                        v->output_queue_tail = NULL;
+                v->n_output_queue--;
+
+                varlink_json_queue_item_free(q);
+        }
+
+        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;
+
+        assert(v);
+        assert(m);
+
+        /* If there are no file descriptors to be queued and no queue entries yet we can shortcut things and
+         * append this entry directly to the output buffer */
+        if (v->n_pushed_fds == 0 && !v->output_queue)
+                return varlink_format_json(v, m);
+
+        if (v->n_output_queue >= VARLINK_QUEUE_MAX)
+                return -ENOBUFS;
+
+        /* Otherwise add a queue entry for this */
+        q = varlink_json_queue_item_new(m, v->pushed_fds, v->n_pushed_fds);
+        if (!q)
+                return -ENOMEM;
+
+        v->n_pushed_fds = 0; /* fds now belong to the queue entry */
+
+        /* We already checked the precondition ourselves so this call cannot fail. */
+        assert_se(varlink_enqueue_item(v, q) >= 0);
+
+        return 0;
+}
+
 static int varlink_dispatch_method(sd_varlink *v) {
         _cleanup_(sd_json_variant_unrefp) sd_json_variant *parameters = NULL;
         sd_varlink_method_flags_t flags = 0;
@@ -1401,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));
@@ -1898,141 +2082,6 @@ _public_ sd_varlink* sd_varlink_flush_close_unref(sd_varlink *v) {
         return sd_varlink_close_unref(v);
 }
 
-static int varlink_format_json(sd_varlink *v, sd_json_variant *m) {
-        _cleanup_(erase_and_freep) char *text = NULL;
-        int sz, r;
-
-        assert(v);
-        assert(m);
-
-        sz = sd_json_variant_format(m, /* flags= */ 0, &text);
-        if (sz < 0)
-                return sz;
-        assert(text[sz] == '\0');
-
-        if (v->output_buffer_size + sz + 1 > VARLINK_BUFFER_MAX)
-                return -ENOBUFS;
-
-        if (DEBUG_LOGGING) {
-                _cleanup_(erase_and_freep) char *censored_text = NULL;
-
-                /* Suppress sensitive fields in the debug output */
-                r = sd_json_variant_format(m, SD_JSON_FORMAT_CENSOR_SENSITIVE, &censored_text);
-                if (r < 0)
-                        return r;
-
-                varlink_log(v, "Sending message: %s", censored_text);
-        }
-
-        if (v->output_buffer_size == 0) {
-
-                free_and_replace(v->output_buffer, text);
-
-                v->output_buffer_size = sz + 1;
-                v->output_buffer_index = 0;
-
-        } else if (v->output_buffer_index == 0) {
-
-                if (!GREEDY_REALLOC(v->output_buffer, v->output_buffer_size + sz + 1))
-                        return -ENOMEM;
-
-                memcpy(v->output_buffer + v->output_buffer_size, text, sz + 1);
-                v->output_buffer_size += sz + 1;
-        } else {
-                char *n;
-                const size_t new_size = v->output_buffer_size + sz + 1;
-
-                n = new(char, new_size);
-                if (!n)
-                        return -ENOMEM;
-
-                memcpy(mempcpy(n, v->output_buffer + v->output_buffer_index, v->output_buffer_size), text, sz + 1);
-
-                free_and_replace(v->output_buffer, n);
-                v->output_buffer_size = new_size;
-                v->output_buffer_index = 0;
-        }
-
-        if (sd_json_variant_is_sensitive_recursive(m))
-                v->output_buffer_sensitive = true; /* Propagate sensitive flag */
-        else
-                text = mfree(text); /* No point in the erase_and_free() destructor declared above */
-
-        return 0;
-}
-
-static int varlink_enqueue_json(sd_varlink *v, sd_json_variant *m) {
-        VarlinkJsonQueueItem *q;
-
-        assert(v);
-        assert(m);
-
-        /* If there are no file descriptors to be queued and no queue entries yet we can shortcut things and
-         * append this entry directly to the output buffer */
-        if (v->n_pushed_fds == 0 && !v->output_queue)
-                return varlink_format_json(v, m);
-
-        if (v->n_output_queue >= VARLINK_QUEUE_MAX)
-                return -ENOBUFS;
-
-        /* Otherwise add a queue entry for this */
-        q = varlink_json_queue_item_new(m, v->pushed_fds, v->n_pushed_fds);
-        if (!q)
-                return -ENOMEM;
-
-        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++;
-        return 0;
-}
-
-static int varlink_format_queue(sd_varlink *v) {
-        int r;
-
-        assert(v);
-
-        /* Takes entries out of the output queue and formats them into the output buffer. But only if this
-         * would not corrupt our fd message boundaries */
-
-        while (v->output_queue) {
-                _cleanup_free_ int *array = NULL;
-
-                assert(v->n_output_queue > 0);
-
-                VarlinkJsonQueueItem *q = v->output_queue;
-
-                if (v->n_output_fds > 0) /* unwritten fds? if we'd add more we'd corrupt the fd message boundaries, hence wait */
-                        return 0;
-
-                if (q->n_fds > 0) {
-                        array = newdup(int, q->fds, q->n_fds);
-                        if (!array)
-                                return -ENOMEM;
-                }
-
-                r = varlink_format_json(v, q->data);
-                if (r < 0)
-                        return r;
-
-                /* Take possession of the queue element's fds */
-                free(v->output_fds);
-                v->output_fds = TAKE_PTR(array);
-                v->n_output_fds = q->n_fds;
-                q->n_fds = 0;
-
-                LIST_REMOVE(queue, v->output_queue, q);
-                if (!v->output_queue)
-                        v->output_queue_tail = NULL;
-                v->n_output_queue--;
-
-                varlink_json_queue_item_free(q);
-        }
-
-        return 0;
-}
-
 _public_ int sd_varlink_send(sd_varlink *v, const char *method, sd_json_variant *parameters) {
         _cleanup_(sd_json_variant_unrefp) sd_json_variant *m = NULL;
         int r;
@@ -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;

src/machine/machined-varlink.c

@@ -426,7 +426,7 @@ static int json_build_local_addresses(const struct local_address *addresses, siz
         return 0;
 }
 
-static int list_machine_one_and_maybe_read_metadata(sd_varlink *link, Machine *m, bool more, AcquireMetadata am) {
+static int list_machine_one_and_maybe_read_metadata(sd_varlink *link, Machine *m, AcquireMetadata am) {
         _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL, *addr_array = NULL;
         _cleanup_strv_free_ char **os_release = NULL;
         uid_t shift = UID_INVALID;
@@ -496,9 +496,6 @@ static int list_machine_one_and_maybe_read_metadata(sd_varlink *link, Machine *m
         if (r < 0)
                 return r;
 
-        if (more)
-                return sd_varlink_notify(link, v);
-
         return sd_varlink_reply(link, v);
 }
 
@@ -528,8 +525,6 @@ static int vl_method_list(sd_varlink *link, sd_json_variant *parameters, sd_varl
         _cleanup_(machine_lookup_parameters_done) MachineLookupParameters p = {
                 .pidref = PIDREF_NULL,
         };
-
-        Machine *machine;
         int r;
 
         assert(link);
@@ -539,34 +534,32 @@ static int vl_method_list(sd_varlink *link, sd_json_variant *parameters, sd_varl
         if (r != 0)
                 return r;
 
+        r = varlink_set_sentinel(link, VARLINK_ERROR_MACHINE_NO_SUCH_MACHINE);
+        if (r < 0)
+                return r;
+
         if (p.name || pidref_is_set(&p.pidref) || pidref_is_automatic(&p.pidref)) {
+                Machine *machine;
                 r = lookup_machine_by_name_or_pidref(link, m, p.name, &p.pidref, &machine);
                 if (r == -ESRCH)
-                        return sd_varlink_error(link, VARLINK_ERROR_MACHINE_NO_SUCH_MACHINE, NULL);
+                        return 0;
                 if (r < 0)
                         return r;
 
-                return list_machine_one_and_maybe_read_metadata(link, machine, /* more= */ false, p.acquire_metadata);
+                return list_machine_one_and_maybe_read_metadata(link, machine, p.acquire_metadata);
         }
 
         if (!FLAGS_SET(flags, SD_VARLINK_METHOD_MORE))
                 return sd_varlink_error(link, SD_VARLINK_ERROR_EXPECTED_MORE, NULL);
 
-        Machine *previous = NULL, *i;
-        HASHMAP_FOREACH(i, m->machines) {
-                if (previous) {
-                        r = list_machine_one_and_maybe_read_metadata(link, previous, /* more= */ true, p.acquire_metadata);
-                        if (r < 0)
-                                return r;
-                }
-
-                previous = i;
+        Machine *machine;
+        HASHMAP_FOREACH(machine, m->machines) {
+                r = list_machine_one_and_maybe_read_metadata(link, machine, p.acquire_metadata);
+                if (r < 0)
+                        return r;
         }
 
-        if (previous)
-                return list_machine_one_and_maybe_read_metadata(link, previous, /* more= */ false, p.acquire_metadata);
-
-        return sd_varlink_error(link, VARLINK_ERROR_MACHINE_NO_SUCH_MACHINE, NULL);
+        return 0;
 }
 
 static int lookup_machine_and_call_method(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata, sd_varlink_method_t method) {
@@ -619,7 +612,7 @@ static int vl_method_open_root_directory(sd_varlink *link, sd_json_variant *para
         return lookup_machine_and_call_method(link, parameters, flags, userdata, vl_method_open_root_directory_internal);
 }
 
-static int list_image_one_and_maybe_read_metadata(Manager *m, sd_varlink *link, Image *image, bool more, AcquireMetadata am) {
+static int list_image_one_and_maybe_read_metadata(Manager *m, sd_varlink *link, Image *image, AcquireMetadata am) {
         int r;
 
         assert(m);
@@ -663,9 +656,6 @@ static int list_image_one_and_maybe_read_metadata(Manager *m, sd_varlink *link,
                         return r;
         }
 
-        if (more)
-                return sd_varlink_notify(link, v);
-
         return sd_varlink_reply(link, v);
 }
 
@@ -691,6 +681,10 @@ static int vl_method_list_images(sd_varlink *link, sd_json_variant *parameters,
         if (r != 0)
                 return r;
 
+        r = varlink_set_sentinel(link, VARLINK_ERROR_MACHINE_IMAGE_NO_SUCH_IMAGE);
+        if (r < 0)
+                return r;
+
         if (p.image_name) {
                 _cleanup_(image_unrefp) Image *found = NULL;
 
@@ -699,11 +693,11 @@ static int vl_method_list_images(sd_varlink *link, sd_json_variant *parameters,
 
                 r = image_find(m->runtime_scope, IMAGE_MACHINE, p.image_name, /* root= */ NULL, &found);
                 if (r == -ENOENT)
-                        return sd_varlink_error(link, VARLINK_ERROR_MACHINE_IMAGE_NO_SUCH_IMAGE, NULL);
+                        return 0;
                 if (r < 0)
                         return log_debug_errno(r, "Failed to find image: %m");
 
-                return list_image_one_and_maybe_read_metadata(m, link, found, /* more= */ false, p.acquire_metadata);
+                return list_image_one_and_maybe_read_metadata(m, link, found, p.acquire_metadata);
         }
 
         if (!FLAGS_SET(flags, SD_VARLINK_METHOD_MORE))
@@ -714,21 +708,14 @@ static int vl_method_list_images(sd_varlink *link, sd_json_variant *parameters,
         if (r < 0)
                 return log_debug_errno(r, "Failed to discover images: %m");
 
-        Image *image, *previous = NULL;
+        Image *image;
         HASHMAP_FOREACH(image, images) {
-                if (previous) {
-                        r = list_image_one_and_maybe_read_metadata(m, link, previous, /* more= */ true, p.acquire_metadata);
-                        if (r < 0)
-                                return r;
-                }
-
-                previous = image;
+                r = list_image_one_and_maybe_read_metadata(m, link, image, p.acquire_metadata);
+                if (r < 0)
+                        return r;
         }
 
-        if (previous)
-                return list_image_one_and_maybe_read_metadata(m, link, previous, /* more= */ false, p.acquire_metadata);
-
-        return sd_varlink_error(link, VARLINK_ERROR_MACHINE_IMAGE_NO_SUCH_IMAGE, NULL);
+        return 0;
 }
 
 static int manager_varlink_init_userdb(Manager *m) {

src/pcrlock/pcrlock.c

@@ -5417,24 +5417,23 @@ static int vl_method_read_event_log(sd_varlink *link, sd_json_variant *parameter
         if (r < 0)
                 return r;
 
-        _cleanup_(sd_json_variant_unrefp) sd_json_variant *rec_cel = NULL;
+        // FIXME: We can't use a NULL sentinel here because the output fields in the IDL are non-nullable.
+        r = varlink_set_sentinel(link, NULL);
+        if (r < 0)
+                return r;
 
         FOREACH_ARRAY(rr, el->records, el->n_records) {
-
-                if (rec_cel) {
-                        r = sd_varlink_notifybo(link, SD_JSON_BUILD_PAIR_VARIANT("record", rec_cel));
-                        if (r < 0)
-                                return r;
-
-                        rec_cel = sd_json_variant_unref(rec_cel);
-                }
-
+                _cleanup_(sd_json_variant_unrefp) sd_json_variant *rec_cel = NULL;
                 r = event_log_record_to_cel(*rr, &recnum, &rec_cel);
                 if (r < 0)
                         return r;
+
+                r = sd_varlink_replybo(link, SD_JSON_BUILD_PAIR_VARIANT("record", rec_cel));
+                if (r < 0)
+                        return r;
         }
 
-        return sd_varlink_replybo(link, SD_JSON_BUILD_PAIR_CONDITION(!!rec_cel, "record", SD_JSON_BUILD_VARIANT(rec_cel)));
+        return 0;
 }
 
 typedef struct MethodMakePolicyParameters {

src/repart/repart.c

@@ -10346,21 +10346,12 @@ static int vl_method_list_candidate_devices(
         if (r < 0)
                 return r;
 
-        if (n == 0)
-                return sd_varlink_error(link, "io.systemd.Repart.NoCandidateDevices", NULL);
+        r = varlink_set_sentinel(link, "io.systemd.Repart.NoCandidateDevices");
+        if (r < 0)
+                return r;
 
-        _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL;
         FOREACH_ARRAY(d, l, n) {
-                if (v) {
-                        r = sd_varlink_notify(link, v);
-                        if (r < 0)
-                                return r;
-
-                        v = sd_json_variant_unref(v);
-                }
-
-                r = sd_json_buildo(
-                                &v,
+                r = sd_varlink_replybo(link,
                                 SD_JSON_BUILD_PAIR_STRING("node", d->node),
                                 JSON_BUILD_PAIR_STRV_NON_EMPTY("symlinks", d->symlinks),
                                 JSON_BUILD_PAIR_UNSIGNED_NOT_EQUAL("diskseq", d->diskseq, UINT64_MAX),
@@ -10372,8 +10363,7 @@ static int vl_method_list_candidate_devices(
                         return r;
         }
 
-        assert(v);
-        return sd_varlink_reply(link, v);
+        return 0;
 }
 
 static JSON_DISPATCH_ENUM_DEFINE(json_dispatch_empty_mode, EmptyMode, empty_mode_from_string);

src/resolve/resolved-manager.c

@@ -886,6 +886,9 @@ Manager* manager_free(Manager *m) {
         manager_dns_stub_stop(m);
         manager_varlink_done(m);
 
+        set_free(m->varlink_query_results_subscription);
+        set_free(m->varlink_dns_configuration_subscription);
+
         manager_socket_graveyard_clear(m);
 
         ordered_set_free(m->dns_extra_stub_listeners);

src/shared/metrics.c

@@ -4,16 +4,9 @@
 #include "log.h"
 #include "metrics.h"
 #include "string-table.h"
-#include "strv.h"
 #include "varlink-io.systemd.Metrics.h"
 #include "varlink-util.h"
 
-static void metric_family_context_done(MetricFamilyContext *ctx) {
-        assert(ctx);
-
-        sd_json_variant_unref(ctx->previous);
-}
-
 int metrics_setup_varlink_server(
                 sd_varlink_server **server, /* in and out param */
                 sd_varlink_server_flags_t flags,
@@ -54,7 +47,7 @@ int metrics_setup_varlink_server(
 
         r = sd_varlink_server_attach_event(s, event, SD_EVENT_PRIORITY_NORMAL);
         if (r < 0)
-                return log_debug_errno(r, "Failed to attach varlink metrics connection to event loop: %m");
+                return log_debug_errno(r, "Failed to attach varlink metrics server to event loop: %m");
 
         *server = TAKE_PTR(s);
 
@@ -69,25 +62,14 @@ static const char * const metric_family_type_table[_METRIC_FAMILY_TYPE_MAX] = {
 
 DEFINE_STRING_TABLE_LOOKUP_TO_STRING(metric_family_type, MetricFamilyType);
 
-static int metric_family_build_send(sd_varlink *link, const MetricFamily *mf, bool more) {
-        _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL;
-        int r;
-
-        assert(link);
+static int metric_family_build_json(const MetricFamily *mf, sd_json_variant **ret) {
         assert(mf);
 
-        r = sd_json_buildo(
-                        &v,
+        return sd_json_buildo(
+                        ret,
                         SD_JSON_BUILD_PAIR_STRING("name", mf->name),
                         SD_JSON_BUILD_PAIR_STRING("description", mf->description),
                         SD_JSON_BUILD_PAIR_STRING("type", metric_family_type_to_string(mf->type)));
-        if (r < 0)
-                return r;
-
-        if (more)
-                return sd_varlink_notify(link, v);
-
-        return sd_varlink_reply(link, v);
 }
 
 int metrics_method_describe(
@@ -110,24 +92,21 @@ int metrics_method_describe(
         if (!FLAGS_SET(flags, SD_VARLINK_METHOD_MORE))
                 return sd_varlink_error(link, SD_VARLINK_ERROR_EXPECTED_MORE, NULL);
 
-        const MetricFamily *previous = NULL;
-        for (const MetricFamily *mf = metric_family_table; mf && mf->name; mf++) {
-                if (previous) {
-                        r = metric_family_build_send(link, previous, /* more= */ true);
-                        if (r < 0)
-                                return log_debug_errno(
-                                                r, "Failed to describe metric family '%s': %m", previous->name);
-                }
-
-                previous = mf;
-        }
-
-        if (!previous)
-                return sd_varlink_error(link, "io.systemd.Metrics.NoSuchMetric", NULL);
-
-        r = metric_family_build_send(link, previous, /* more= */ false);
+        r = varlink_set_sentinel(link, "io.systemd.Metrics.NoSuchMetric");
         if (r < 0)
-                return log_debug_errno(r, "Failed to describe metric family '%s': %m", previous->name);
+                return r;
+
+        for (const MetricFamily *mf = metric_family_table; mf && mf->name; mf++) {
+                _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL;
+
+                r = metric_family_build_json(mf, &v);
+                if (r < 0)
+                        return log_debug_errno(r, "Failed to describe metric family '%s': %m", mf->name);
+
+                r = sd_varlink_reply(link, v);
+                if (r < 0)
+                        return log_debug_errno(r, "Failed to send varlink reply: %m");
+        }
 
         return 0;
 }
@@ -152,100 +131,47 @@ int metrics_method_list(
         if (!FLAGS_SET(flags, SD_VARLINK_METHOD_MORE))
                 return sd_varlink_error(link, SD_VARLINK_ERROR_EXPECTED_MORE, NULL);
 
-        _cleanup_(metric_family_context_done) MetricFamilyContext ctx = { .link = link };
+        r = varlink_set_sentinel(link, "io.systemd.Metrics.NoSuchMetric");
+        if (r < 0)
+                return r;
+
+        MetricFamilyContext ctx = { .link = link };
         for (const MetricFamily *mf = metric_family_table; mf && mf->name; mf++) {
-                assert(mf->generate_cb);
+                assert(mf->generate);
 
                 ctx.metric_family = mf;
-                r = mf->generate_cb(&ctx, userdata);
+                r = mf->generate(&ctx, userdata);
                 if (r < 0)
                         return log_debug_errno(
                                         r, "Failed to list metrics for metric family '%s': %m", mf->name);
         }
 
-        if (!ctx.previous)
-                return sd_varlink_error(link, "io.systemd.Metrics.NoSuchMetric", NULL);
-
-        /* produce the last metric */
-        return sd_varlink_reply(link, ctx.previous);
+        return 0;
 }
 
-static int metric_set_fields(sd_json_variant **v, char **field_pairs) {
-        _cleanup_(sd_json_variant_unrefp) sd_json_variant *w = NULL;
-        size_t n;
-        int r;
-
-        assert(v);
-
-        n = strv_length(field_pairs);
-        if (n == 0)
-                return 0;
-
-        if (n % 2 != 0)
-                return log_debug_errno(SYNTHETIC_ERRNO(ERANGE), "Odd number of field pairs: %zu", n);
-
-        sd_json_variant **array = new0(sd_json_variant *, n);
-        if (!array)
-                return log_oom();
-
-        CLEANUP_ARRAY(array, n, sd_json_variant_unref_many);
-
-        int i = 0;
-        STRV_FOREACH_PAIR(key, value, field_pairs) {
-                r = sd_json_variant_new_string(&array[i++], *key);
-                if (r < 0)
-                        return log_debug_errno(r, "Failed to create key variant: %m");
-
-                r = sd_json_variant_new_string(&array[i++], *value);
-                if (r < 0)
-                        return log_debug_errno(r, "Failed to create value variant: %m");
-        }
-
-        r = sd_json_variant_new_object(&w, array, n);
-        if (r < 0)
-                return log_debug_errno(r, "Failed to allocate JSON object: %m");
-
-        return sd_json_variant_set_field(v, "fields", w);
-}
-
-static int metric_build_send(MetricFamilyContext *context, const char *object, sd_json_variant *value, char **field_pairs) {
-        _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL;
-        int r;
-
+static int metric_build_send(MetricFamilyContext *context, const char *object, sd_json_variant *value, sd_json_variant *fields) {
         assert(context);
         assert(value);
         assert(context->link);
         assert(context->metric_family);
 
-        r = sd_json_buildo(
-                        &v,
+        if (fields) {
+                assert(sd_json_variant_is_object(fields));
+
+                _unused_ const char *k;
+                _unused_ sd_json_variant *e;
+                JSON_VARIANT_OBJECT_FOREACH(k, e, fields)
+                        assert(sd_json_variant_is_string(e));
+        }
+
+        return sd_varlink_replybo(context->link,
                         SD_JSON_BUILD_PAIR_STRING("name", context->metric_family->name),
                         JSON_BUILD_PAIR_STRING_NON_EMPTY("object", object),
-                        SD_JSON_BUILD_PAIR("value", SD_JSON_BUILD_VARIANT(value)));
-        /* TODO JSON_BUILD_PAIR_OBJECT_STRV_NOT_NULL */
-        if (r < 0)
-                return r;
-
-        if (field_pairs) { /* NULL => no fields object, empty strv => fields:{} */
-                r = metric_set_fields(&v, field_pairs);
-                if (r < 0)
-                        return r;
-        }
-
-        if (context->previous) {
-                r = sd_varlink_notify(context->link, context->previous);
-                if (r < 0)
-                        return r;
-
-                context->previous = sd_json_variant_unref(context->previous);
-        }
-
-        context->previous = TAKE_PTR(v);
-
-        return 0;
+                        SD_JSON_BUILD_PAIR("value", SD_JSON_BUILD_VARIANT(value)),
+                        JSON_BUILD_PAIR_VARIANT_NON_NULL("fields", fields));
 }
 
-int metric_build_send_string(MetricFamilyContext *context, const char *object, const char *value, char **field_pairs) {
+int metric_build_send_string(MetricFamilyContext *context, const char *object, const char *value, sd_json_variant *fields) {
         _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL;
         int r;
 
@@ -255,10 +181,10 @@ int metric_build_send_string(MetricFamilyContext *context, const char *object, c
         if (r < 0)
                 return log_debug_errno(r, "Failed to allocate JSON string: %m");
 
-        return metric_build_send(context, object, v, field_pairs);
+        return metric_build_send(context, object, v, fields);
 }
 
-int metric_build_send_unsigned(MetricFamilyContext *context, const char *object, uint64_t value, char **field_pairs) {
+int metric_build_send_unsigned(MetricFamilyContext *context, const char *object, uint64_t value, sd_json_variant *fields) {
         _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL;
         int r;
 
@@ -266,5 +192,5 @@ int metric_build_send_unsigned(MetricFamilyContext *context, const char *object,
         if (r < 0)
                 return log_debug_errno(r, "Failed to allocate JSON unsigned: %m");
 
-        return metric_build_send(context, object, v, field_pairs);
+        return metric_build_send(context, object, v, fields);
 }

src/shared/metrics.h

@@ -1,9 +1,7 @@
 /* SPDX-License-Identifier: LGPL-2.1-or-later */
 #pragma once
 
-#include "sd-varlink.h"
-
-#include "macro-fundamental.h"
+#include "shared-forward.h"
 
 typedef enum MetricFamilyType {
         METRIC_FAMILY_TYPE_COUNTER,
@@ -18,16 +16,15 @@ typedef struct MetricFamily MetricFamily;
 typedef struct MetricFamilyContext {
         const MetricFamily* metric_family;
         sd_varlink *link;
-        sd_json_variant *previous;
 } MetricFamilyContext;
 
-typedef int (*metric_family_generate_cb_t) (MetricFamilyContext *mfc, void *userdata);
+typedef int (*metric_family_generate_func_t) (MetricFamilyContext *mfc, void *userdata);
 
 typedef struct MetricFamily {
         const char *name;
         const char *description;
         MetricFamilyType type;
-        metric_family_generate_cb_t generate_cb;
+        metric_family_generate_func_t generate;
 } MetricFamily;
 
 int metrics_setup_varlink_server(
@@ -38,9 +35,10 @@ int metrics_setup_varlink_server(
                 sd_varlink_method_t vl_method_describe_cb,
                 void *userdata);
 
-const char* metric_family_type_to_string(MetricFamilyType i) _const_;
+DECLARE_STRING_TABLE_LOOKUP_TO_STRING(metric_family_type, MetricFamilyType);
+
 int metrics_method_describe(const MetricFamily metric_family_table[], sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata);
 int metrics_method_list(const MetricFamily metric_family_table[], sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata);
 
-int metric_build_send_string(MetricFamilyContext* context, const char *object, const char *value, char **field_pairs);
-int metric_build_send_unsigned(MetricFamilyContext* context, const char *object, uint64_t value, char **field_pairs);
+int metric_build_send_string(MetricFamilyContext* context, const char *object, const char *value, sd_json_variant *fields);
+int metric_build_send_unsigned(MetricFamilyContext* context, const char *object, uint64_t value, sd_json_variant *fields);

src/shared/varlink-io.systemd.BootControl.c

@@ -87,7 +87,7 @@ static SD_VARLINK_DEFINE_METHOD_FULL(
                 ListBootEntries,
                 SD_VARLINK_REQUIRES_MORE,
                 SD_VARLINK_FIELD_COMMENT("A boot menu entry structure"),
-                SD_VARLINK_DEFINE_OUTPUT_BY_TYPE(entry, BootEntry, SD_VARLINK_NULLABLE));
+                SD_VARLINK_DEFINE_OUTPUT_BY_TYPE(entry, BootEntry, 0));
 
 static SD_VARLINK_DEFINE_METHOD(
                 SetRebootToFirmware,

src/sysext/sysext.c

@@ -2745,7 +2745,6 @@ static int vl_method_list(sd_varlink *link, sd_json_variant *parameters, sd_varl
                 VARLINK_DISPATCH_POLKIT_FIELD,
                 {}
         };
-        _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL;
         Hashmap **polkit_registry = ASSERT_PTR(userdata);
         int r;
 
@@ -2775,26 +2774,23 @@ static int vl_method_list(sd_varlink *link, sd_json_variant *parameters, sd_varl
         if (r < 0)
                 return r;
 
+        r = varlink_set_sentinel(link, "io.systemd.sysext.NoImagesFound");
+        if (r < 0)
+                return r;
+
         Image *img;
         HASHMAP_FOREACH(img, images) {
-                if (v) {
-                        /* Send previous item with more=true */
-                        r = sd_varlink_notify(link, v);
-                        if (r < 0)
-                                return r;
-                }
-
-                v = sd_json_variant_unref(v);
-
+                _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL;
                 r = image_to_json(img, &v);
                 if (r < 0)
                         return r;
+
+                r = sd_varlink_reply(link, v);
+                if (r < 0)
+                        return r;
         }
 
-        if (v)  /* Send final item with more=false */
-                return sd_varlink_reply(link, v);
-
-        return sd_varlink_error(link, "io.systemd.sysext.NoImagesFound", NULL);
+        return 0;
 }
 
 static int verb_help(int argc, char **argv, void *userdata) {

src/test/test-varlink.c

@@ -441,4 +441,315 @@ TEST(invalid_parameter) {
         ASSERT_OK(sd_event_loop(e));
 }
 
+static int method_with_error_sentinel(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata) {
+        /* Set an error sentinel and return without sending a reply. The sentinel error should be sent automatically. */
+        ASSERT_OK(varlink_set_sentinel(link, "io.test.SentinelError"));
+        return 0;
+}
+
+static int reply_sentinel_error(sd_varlink *link, sd_json_variant *parameters, const char *error_id, sd_varlink_reply_flags_t flags, void *userdata) {
+        ASSERT_STREQ(error_id, "io.test.SentinelError");
+        ASSERT_OK(sd_event_exit(sd_varlink_get_event(link), EXIT_SUCCESS));
+        return 0;
+}
+
+TEST(sentinel_error) {
+        _cleanup_(sd_event_unrefp) sd_event *e = NULL;
+        ASSERT_OK(sd_event_default(&e));
+
+        _cleanup_(sd_varlink_server_unrefp) sd_varlink_server *s = NULL;
+        ASSERT_OK(sd_varlink_server_new(&s, 0));
+
+        ASSERT_OK(sd_varlink_server_attach_event(s, e, 0));
+
+        ASSERT_OK(sd_varlink_server_bind_method(s, "io.test.ErrorSentinel", method_with_error_sentinel));
+
+        int connfd[2];
+        ASSERT_OK_ERRNO(socketpair(AF_UNIX, SOCK_STREAM|SOCK_NONBLOCK|SOCK_CLOEXEC, 0, connfd));
+        ASSERT_OK(sd_varlink_server_add_connection(s, connfd[0], /* ret= */ NULL));
+
+        _cleanup_(sd_varlink_unrefp) sd_varlink *c = NULL;
+        ASSERT_OK(sd_varlink_connect_fd(&c, connfd[1]));
+
+        ASSERT_OK(sd_varlink_attach_event(c, e, 0));
+
+        ASSERT_OK(sd_varlink_bind_reply(c, reply_sentinel_error));
+
+        ASSERT_OK(sd_varlink_invoke(c, "io.test.ErrorSentinel", /* parameters= */ NULL));
+
+        ASSERT_OK(sd_event_loop(e));
+}
+
+static int method_with_empty_sentinel(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata) {
+        /* Set an empty sentinel and return without sending a reply. An empty reply should be sent automatically. */
+        ASSERT_OK(varlink_set_sentinel(link, /* error_id= */ NULL));
+        return 0;
+}
+
+static int reply_sentinel_empty(sd_varlink *link, sd_json_variant *parameters, const char *error_id, sd_varlink_reply_flags_t flags, void *userdata) {
+        ASSERT_NULL(error_id);
+        ASSERT_TRUE(sd_json_variant_is_blank_object(parameters));
+        ASSERT_OK(sd_event_exit(sd_varlink_get_event(link), EXIT_SUCCESS));
+        return 0;
+}
+
+TEST(sentinel_empty) {
+        _cleanup_(sd_event_unrefp) sd_event *e = NULL;
+        ASSERT_OK(sd_event_default(&e));
+
+        _cleanup_(sd_varlink_server_unrefp) sd_varlink_server *s = NULL;
+        ASSERT_OK(sd_varlink_server_new(&s, 0));
+
+        ASSERT_OK(sd_varlink_server_attach_event(s, e, 0));
+
+        ASSERT_OK(sd_varlink_server_bind_method(s, "io.test.EmptySentinel", method_with_empty_sentinel));
+
+        int connfd[2];
+        ASSERT_OK_ERRNO(socketpair(AF_UNIX, SOCK_STREAM|SOCK_NONBLOCK|SOCK_CLOEXEC, 0, connfd));
+        ASSERT_OK(sd_varlink_server_add_connection(s, connfd[0], /* ret= */ NULL));
+
+        _cleanup_(sd_varlink_unrefp) sd_varlink *c = NULL;
+        ASSERT_OK(sd_varlink_connect_fd(&c, connfd[1]));
+
+        ASSERT_OK(sd_varlink_attach_event(c, e, 0));
+
+        ASSERT_OK(sd_varlink_bind_reply(c, reply_sentinel_empty));
+
+        ASSERT_OK(sd_varlink_invoke(c, "io.test.EmptySentinel", /* parameters= */ NULL));
+
+        ASSERT_OK(sd_event_loop(e));
+}
+
+static int method_with_sentinel_but_reply(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata) {
+        /* Set a sentinel but also send a reply. The sentinel should not be used. */
+        ASSERT_OK(varlink_set_sentinel(link, "io.test.SentinelError"));
+        return sd_varlink_replybo(link, SD_JSON_BUILD_PAIR_STRING("result", "explicit-reply"));
+}
+
+static int reply_sentinel_explicit(sd_varlink *link, sd_json_variant *parameters, const char *error_id, sd_varlink_reply_flags_t flags, void *userdata) {
+        ASSERT_NULL(error_id);
+        ASSERT_STREQ(sd_json_variant_string(sd_json_variant_by_key(parameters, "result")), "explicit-reply");
+        ASSERT_OK(sd_event_exit(sd_varlink_get_event(link), EXIT_SUCCESS));
+        return 0;
+}
+
+TEST(sentinel_with_explicit_reply) {
+        _cleanup_(sd_event_unrefp) sd_event *e = NULL;
+        ASSERT_OK(sd_event_default(&e));
+
+        _cleanup_(sd_varlink_server_unrefp) sd_varlink_server *s = NULL;
+        ASSERT_OK(sd_varlink_server_new(&s, 0));
+
+        ASSERT_OK(sd_varlink_server_attach_event(s, e, 0));
+
+        ASSERT_OK(sd_varlink_server_bind_method(s, "io.test.SentinelButReply", method_with_sentinel_but_reply));
+
+        int connfd[2];
+        ASSERT_OK_ERRNO(socketpair(AF_UNIX, SOCK_STREAM|SOCK_NONBLOCK|SOCK_CLOEXEC, 0, connfd));
+        ASSERT_OK(sd_varlink_server_add_connection(s, connfd[0], /* ret= */ NULL));
+
+        _cleanup_(sd_varlink_unrefp) sd_varlink *c = NULL;
+        ASSERT_OK(sd_varlink_connect_fd(&c, connfd[1]));
+
+        ASSERT_OK(sd_varlink_attach_event(c, e, 0));
+
+        ASSERT_OK(sd_varlink_bind_reply(c, reply_sentinel_explicit));
+
+        ASSERT_OK(sd_varlink_invoke(c, "io.test.SentinelButReply", /* parameters= */ NULL));
+
+        ASSERT_OK(sd_event_loop(e));
+}
+
+static int method_with_oneway_sentinel(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata) {
+        /* The method was called oneway, so 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(varlink_set_sentinel(link, "io.test.SentinelError"));
+        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"));
+}
+
+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) {
+        /* If we get here, it means the oneway sentinel call didn't break the connection and the server
+         * 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(sd_event_exit(sd_varlink_get_event(link), EXIT_SUCCESS));
+        return 0;
+}
+
+TEST(sentinel_oneway) {
+        _cleanup_(sd_event_unrefp) sd_event *e = NULL;
+        ASSERT_OK(sd_event_default(&e));
+
+        _cleanup_(sd_varlink_server_unrefp) sd_varlink_server *s = NULL;
+        ASSERT_OK(sd_varlink_server_new(&s, 0));
+
+        ASSERT_OK(sd_varlink_server_attach_event(s, e, 0));
+
+        ASSERT_OK(sd_varlink_server_bind_method(s, "io.test.OnewaySentinel", method_with_oneway_sentinel));
+        ASSERT_OK(sd_varlink_server_bind_method(s, "io.test.Pong", method_oneway_sentinel_pong));
+
+        int connfd[2];
+        ASSERT_OK_ERRNO(socketpair(AF_UNIX, SOCK_STREAM|SOCK_NONBLOCK|SOCK_CLOEXEC, 0, connfd));
+        ASSERT_OK(sd_varlink_server_add_connection(s, connfd[0], /* ret= */ NULL));
+
+        _cleanup_(sd_varlink_unrefp) sd_varlink *c = NULL;
+        ASSERT_OK(sd_varlink_connect_fd(&c, connfd[1]));
+
+        ASSERT_OK(sd_varlink_attach_event(c, e, 0));
+
+        /* Send a oneway call with a sentinel — the sentinel should be silently ignored. */
+        ASSERT_OK(sd_varlink_send(c, "io.test.OnewaySentinel", /* parameters= */ NULL));
+
+        /* Follow up with a regular call to verify the server is still functional. */
+        ASSERT_OK(sd_varlink_bind_reply(c, reply_oneway_sentinel_pong));
+        ASSERT_OK(sd_varlink_invoke(c, "io.test.Pong", /* parameters= */ NULL));
+
+        ASSERT_OK(sd_event_loop(e));
+}
+
+static int method_with_fd_sentinel(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata) {
+        _cleanup_close_ int fd1 = -EBADF, fd2 = -EBADF;
+
+        ASSERT_TRUE(FLAGS_SET(flags, SD_VARLINK_METHOD_MORE));
+
+        /* 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. */
+        ASSERT_OK(varlink_set_sentinel(link, /* error_id= */ NULL));
+
+        /* First reply: push one fd with "alpha" content */
+        ASSERT_OK(fd1 = memfd_new_and_seal_string("data", "alpha"));
+        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)));
+
+        /* 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)));
+
+        return 0;
+}
+
+static int reply_sentinel_fd(sd_varlink *link, sd_json_variant *parameters, const char *error_id, sd_varlink_reply_flags_t flags, void *userdata) {
+        int *state = ASSERT_PTR(sd_varlink_get_userdata(link));
+
+        if (*state == 0) {
+                /* First reply: should carry "continues" flag and fd with "alpha" */
+                ASSERT_NULL(error_id);
+                ASSERT_TRUE(FLAGS_SET(flags, SD_VARLINK_REPLY_CONTINUES));
+                ASSERT_EQ(sd_json_variant_integer(sd_json_variant_by_key(parameters, "index")), 0);
+
+                int fd;
+                ASSERT_OK(fd = sd_varlink_peek_fd(link, 0));
+                test_fd(fd, "alpha", STRLEN("alpha"));
+                (*state)++;
+        } else if (*state == 1) {
+                /* Second (final) reply: no "continues" flag, fd with "beta" */
+                ASSERT_NULL(error_id);
+                ASSERT_FALSE(FLAGS_SET(flags, SD_VARLINK_REPLY_CONTINUES));
+                ASSERT_EQ(sd_json_variant_integer(sd_json_variant_by_key(parameters, "index")), 1);
+
+                int fd;
+                ASSERT_OK(fd = sd_varlink_peek_fd(link, 0));
+                test_fd(fd, "beta", STRLEN("beta"));
+
+                ASSERT_OK(sd_event_exit(sd_varlink_get_event(link), EXIT_SUCCESS));
+        } else
+                assert_not_reached();
+
+        return 0;
+}
+
+TEST(sentinel_with_fds) {
+        _cleanup_(sd_event_unrefp) sd_event *e = NULL;
+        ASSERT_OK(sd_event_default(&e));
+
+        _cleanup_(sd_varlink_server_unrefp) sd_varlink_server *s = NULL;
+        ASSERT_OK(sd_varlink_server_new(&s, SD_VARLINK_SERVER_ALLOW_FD_PASSING_INPUT|SD_VARLINK_SERVER_ALLOW_FD_PASSING_OUTPUT));
+
+        ASSERT_OK(sd_varlink_server_attach_event(s, e, 0));
+
+        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));
+        ASSERT_OK(sd_varlink_server_add_connection(s, connfd[0], /* ret= */ NULL));
+
+        _cleanup_(sd_varlink_unrefp) sd_varlink *c = NULL;
+        ASSERT_OK(sd_varlink_connect_fd(&c, connfd[1]));
+        ASSERT_OK(sd_varlink_set_allow_fd_passing_input(c, true));
+        ASSERT_OK(sd_varlink_set_allow_fd_passing_output(c, true));
+
+        ASSERT_OK(sd_varlink_attach_event(c, e, 0));
+
+        int state = 0;
+        sd_varlink_set_userdata(c, &state);
+        ASSERT_OK(sd_varlink_bind_reply(c, reply_sentinel_fd));
+
+        ASSERT_OK(sd_varlink_observe(c, "io.test.FDSentinel", /* parameters= */ NULL));
+
+        ASSERT_OK(sd_event_loop(e));
+}
+
+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);
+}
+
+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) {
+        int *state = ASSERT_PTR(sd_varlink_get_userdata(link));
+
+        if (*state == 0) {
+                /* First callback: should be the notify (no error, has "more" flag) */
+                ASSERT_NULL(error_id);
+                ASSERT_TRUE(FLAGS_SET(flags, SD_VARLINK_REPLY_CONTINUES));
+                ASSERT_STREQ(sd_json_variant_string(sd_json_variant_by_key(parameters, "status")), "in-progress");
+                (*state)++;
+        } else if (*state == 1) {
+                /* Second callback: should be the error */
+                ASSERT_STREQ(error_id, "io.test.OperationFailed");
+                ASSERT_FALSE(FLAGS_SET(flags, SD_VARLINK_REPLY_CONTINUES));
+                ASSERT_OK(sd_event_exit(sd_varlink_get_event(link), EXIT_SUCCESS));
+        } else
+                assert_not_reached();
+
+        return 0;
+}
+
+TEST(notify_then_error) {
+        _cleanup_(sd_event_unrefp) sd_event *e = NULL;
+        ASSERT_OK(sd_event_default(&e));
+
+        _cleanup_(sd_varlink_server_unrefp) sd_varlink_server *s = NULL;
+        ASSERT_OK(sd_varlink_server_new(&s, 0));
+
+        ASSERT_OK(sd_varlink_server_attach_event(s, e, 0));
+
+        ASSERT_OK(sd_varlink_server_bind_method(s, "io.test.NotifyThenError", method_with_notify_then_error));
+
+        int connfd[2];
+        ASSERT_OK_ERRNO(socketpair(AF_UNIX, SOCK_STREAM|SOCK_NONBLOCK|SOCK_CLOEXEC, 0, connfd));
+        ASSERT_OK(sd_varlink_server_add_connection(s, connfd[0], /* ret= */ NULL));
+
+        _cleanup_(sd_varlink_unrefp) sd_varlink *c = NULL;
+        ASSERT_OK(sd_varlink_connect_fd(&c, connfd[1]));
+
+        ASSERT_OK(sd_varlink_attach_event(c, e, 0));
+
+        int state = 0;
+        sd_varlink_set_userdata(c, &state);
+        ASSERT_OK(sd_varlink_bind_reply(c, reply_notify_then_error));
+
+        ASSERT_OK(sd_varlink_observe(c, "io.test.NotifyThenError", /* parameters= */ NULL));
+
+        ASSERT_OK(sd_event_loop(e));
+}
+
 DEFINE_TEST_MAIN(LOG_DEBUG);

src/userdb/userwork.c

@@ -153,7 +153,6 @@ static int vl_method_get_user_record(sd_varlink *link, sd_json_variant *paramete
                 {}
         };
 
-        _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL;
         _cleanup_(user_record_unrefp) UserRecord *hr = NULL;
         _cleanup_(lookup_parameters_done) LookupParameters p = {
                 .uid = UID_INVALID,
@@ -173,13 +172,16 @@ static int vl_method_get_user_record(sd_varlink *link, sd_json_variant *paramete
                      * we are done'; == 0 means 'not processed, caller should process now' */
                 return r;
 
+        r = varlink_set_sentinel(link, "io.systemd.UserDatabase.NoRecordFound");
+        if (r < 0)
+                return r;
+
         if (uid_is_valid(p.uid))
                 r = userdb_by_uid(p.uid, &p.match, userdb_flags, &hr);
         else if (p.name)
                 r = userdb_by_name(p.name, &p.match, userdb_flags, &hr);
         else {
                 _cleanup_(userdb_iterator_freep) UserDBIterator *iterator = NULL;
-                _cleanup_(sd_json_variant_unrefp) sd_json_variant *last = NULL;
 
                 r = userdb_all(&p.match, userdb_flags, &iterator);
                 if (IN_SET(r, -ESRCH, -ENOLINK))
@@ -189,7 +191,7 @@ static int vl_method_get_user_record(sd_varlink *link, sd_json_variant *paramete
                          * implementation detail and always return NoRecordFound in this case, since from a
                          * client's perspective it's irrelevant if there was no entry at all or just not on
                          * the service that the query was limited to. */
-                        return sd_varlink_error(link, "io.systemd.UserDatabase.NoRecordFound", NULL);
+                        return 0;
                 if (r < 0)
                         return r;
 
@@ -202,26 +204,20 @@ static int vl_method_get_user_record(sd_varlink *link, sd_json_variant *paramete
                         if (r < 0)
                                 return r;
 
-                        if (last) {
-                                r = sd_varlink_notify(link, last);
-                                if (r < 0)
-                                        return r;
+                        _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL;
+                        r = build_user_json(link, z, &v);
+                        if (r < 0)
+                                return r;
 
-                                last = sd_json_variant_unref(last);
-                        }
-
-                        r = build_user_json(link, z, &last);
+                        r = sd_varlink_reply(link, v);
                         if (r < 0)
                                 return r;
                 }
 
-                if (!last)
-                        return sd_varlink_error(link, "io.systemd.UserDatabase.NoRecordFound", NULL);
-
-                return sd_varlink_reply(link, last);
+                return 0;
         }
         if (r == -ESRCH)
-                return sd_varlink_error(link, "io.systemd.UserDatabase.NoRecordFound", NULL);
+                return 0;
         if (r == -ENOEXEC)
                 return sd_varlink_error(link, "io.systemd.UserDatabase.NonMatchingRecordFound", NULL);
         if (r < 0) {
@@ -233,6 +229,7 @@ static int vl_method_get_user_record(sd_varlink *link, sd_json_variant *paramete
             (p.name && !user_record_matches_user_name(hr, p.name)))
                 return sd_varlink_error(link, "io.systemd.UserDatabase.ConflictingRecordFound", NULL);
 
+        _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL;
         r = build_user_json(link, hr, &v);
         if (r < 0)
                 return r;
@@ -298,7 +295,6 @@ static int vl_method_get_group_record(sd_varlink *link, sd_json_variant *paramet
                 {}
         };
 
-        _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL;
         _cleanup_(group_record_unrefp) GroupRecord *g = NULL;
         _cleanup_(lookup_parameters_done) LookupParameters p = {
                 .gid = GID_INVALID,
@@ -317,17 +313,20 @@ static int vl_method_get_group_record(sd_varlink *link, sd_json_variant *paramet
         if (r != 0)
                 return r;
 
+        r = varlink_set_sentinel(link, "io.systemd.UserDatabase.NoRecordFound");
+        if (r < 0)
+                return r;
+
         if (gid_is_valid(p.gid))
                 r = groupdb_by_gid(p.gid, &p.match, userdb_flags, &g);
         else if (p.name)
                 r = groupdb_by_name(p.name, &p.match, userdb_flags, &g);
         else {
                 _cleanup_(userdb_iterator_freep) UserDBIterator *iterator = NULL;
-                _cleanup_(sd_json_variant_unrefp) sd_json_variant *last = NULL;
 
                 r = groupdb_all(&p.match, userdb_flags, &iterator);
                 if (IN_SET(r, -ESRCH, -ENOLINK))
-                        return sd_varlink_error(link, "io.systemd.UserDatabase.NoRecordFound", NULL);
+                        return 0;
                 if (r < 0)
                         return r;
 
@@ -340,26 +339,20 @@ static int vl_method_get_group_record(sd_varlink *link, sd_json_variant *paramet
                         if (r < 0)
                                 return r;
 
-                        if (last) {
-                                r = sd_varlink_notify(link, last);
-                                if (r < 0)
-                                        return r;
+                        _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL;
+                        r = build_group_json(link, z, &v);
+                        if (r < 0)
+                                return r;
 
-                                last = sd_json_variant_unref(last);
-                        }
-
-                        r = build_group_json(link, z, &last);
+                        r = sd_varlink_reply(link, v);
                         if (r < 0)
                                 return r;
                 }
 
-                if (!last)
-                        return sd_varlink_error(link, "io.systemd.UserDatabase.NoRecordFound", NULL);
-
-                return sd_varlink_reply(link, last);
+                return 0;
         }
         if (r == -ESRCH)
-                return sd_varlink_error(link, "io.systemd.UserDatabase.NoRecordFound", NULL);
+                return 0;
         if (r == -ENOEXEC)
                 return sd_varlink_error(link, "io.systemd.UserDatabase.NonMatchingRecordFound", NULL);
         if (r < 0) {
@@ -371,6 +364,7 @@ static int vl_method_get_group_record(sd_varlink *link, sd_json_variant *paramet
             (p.name && !group_record_matches_group_name(g, p.name)))
                 return sd_varlink_error(link, "io.systemd.UserDatabase.ConflictingRecordFound", NULL);
 
+        _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL;
         r = build_group_json(link, g, &v);
         if (r < 0)
                 return r;
@@ -392,7 +386,6 @@ static int vl_method_get_memberships(sd_varlink *link, sd_json_variant *paramete
                 {}
         };
 
-        _cleanup_free_ char *last_user_name = NULL, *last_group_name = NULL;
         _cleanup_(userdb_iterator_freep) UserDBIterator *iterator = NULL;
         MembershipLookupParameters p = {};
         UserDBFlags userdb_flags;
@@ -408,6 +401,10 @@ static int vl_method_get_memberships(sd_varlink *link, sd_json_variant *paramete
         if (r != 0)
                 return r;
 
+        r = varlink_set_sentinel(link, "io.systemd.UserDatabase.NoRecordFound");
+        if (r < 0)
+                return r;
+
         if (p.group_name)
                 r = membershipdb_by_group(p.group_name, userdb_flags, &iterator);
         else if (p.user_name)
@@ -415,7 +412,7 @@ static int vl_method_get_memberships(sd_varlink *link, sd_json_variant *paramete
         else
                 r = membershipdb_all(userdb_flags, &iterator);
         if (IN_SET(r, -ESRCH, -ENOLINK))
-                return sd_varlink_error(link, "io.systemd.UserDatabase.NoRecordFound", NULL);
+                return 0;
         if (r < 0)
                 return r;
 
@@ -432,32 +429,15 @@ static int vl_method_get_memberships(sd_varlink *link, sd_json_variant *paramete
                 if (p.group_name && p.user_name && !streq(group_name, p.group_name))
                         continue;
 
-                if (last_user_name) {
-                        assert(last_group_name);
-
-                        r = sd_varlink_notifybo(
-                                        link,
-                                        SD_JSON_BUILD_PAIR_STRING("userName", last_user_name),
-                                        SD_JSON_BUILD_PAIR_STRING("groupName", last_group_name));
-                        if (r < 0)
-                                return r;
-                }
-
-                free_and_replace(last_user_name, user_name);
-                free_and_replace(last_group_name, group_name);
+                r = sd_varlink_replybo(
+                                link,
+                                SD_JSON_BUILD_PAIR_STRING("userName", user_name),
+                                SD_JSON_BUILD_PAIR_STRING("groupName", group_name));
+                if (r < 0)
+                        return r;
         }
 
-        if (!last_user_name) {
-                assert(!last_group_name);
-                return sd_varlink_error(link, "io.systemd.UserDatabase.NoRecordFound", NULL);
-        }
-
-        assert(last_group_name);
-
-        return sd_varlink_replybo(
-                        link,
-                        SD_JSON_BUILD_PAIR_STRING("userName", last_user_name),
-                        SD_JSON_BUILD_PAIR_STRING("groupName", last_group_name));
+        return 0;
 }
 
 static int process_connection(sd_varlink_server *server, int _fd) {