sd-bus: when installing a match make sure not to apply it to already queued messages

This tweaks match installation a bit: the match callbacks are now only
called for messages read after the AddMatch() reply was received and
never anything already read before. Thus, installing a match gives you a
time guarantee: only messages received after it will be matched.

This is useful when listening to PropertiesChanged signals as an example
to ensure that only changes after the point the match was installed are
honoured, nothing before.

src/libsystemd/sd-bus/bus-control.c

@@ -803,9 +803,12 @@ _public_ int sd_bus_get_owner_creds(sd_bus *bus, uint64_t mask, sd_bus_creds **r
 
 int bus_add_match_internal(
                 sd_bus *bus,
-                const char *match) {
+                const char *match,
+                uint64_t *ret_counter) {
 
+        _cleanup_(sd_bus_message_unrefp) sd_bus_message *reply = NULL;
         const char *e;
+        int r;
 
         assert(bus);
 
@@ -814,16 +817,24 @@ int bus_add_match_internal(
 
         e = append_eavesdrop(bus, match);
 
-        return sd_bus_call_method(
+        r = sd_bus_call_method(
                         bus,
                         "org.freedesktop.DBus",
                         "/org/freedesktop/DBus",
                         "org.freedesktop.DBus",
                         "AddMatch",
                         NULL,
-                        NULL,
+                        &reply,
                         "s",
                         e);
+        if (r < 0)
+                return r;
+
+        /* If the caller asked for it, return the read counter of the reply */
+        if (ret_counter)
+                *ret_counter = reply->read_counter;
+
+        return r;
 }
 
 int bus_add_match_internal_async(

src/libsystemd/sd-bus/bus-control.h

@@ -3,7 +3,7 @@
 
 #include "sd-bus.h"
 
-int bus_add_match_internal(sd_bus *bus, const char *match);
+int bus_add_match_internal(sd_bus *bus, const char *match, uint64_t *ret_counter);
 int bus_add_match_internal_async(sd_bus *bus, sd_bus_slot **ret, const char *match, sd_bus_message_handler_t callback, void *userdata);
 
 int bus_remove_match_internal(sd_bus *bus, const char *match);

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

@@ -44,6 +44,11 @@ struct match_callback {
 
         unsigned last_iteration;
 
+        /* Don't dispatch this slot with with messages that arrived in any iteration before or at the this
+         * one. We use this to ensure that matches don't apply "retroactively" and thus can confuse the
+         * caller: matches will only match incoming messages from the moment on the match was installed. */
+        uint64_t after;
+
         char *match_string;
 
         struct bus_match_node *match_node;

src/libsystemd/sd-bus/bus-match.c

@@ -287,8 +287,16 @@ int bus_match_run(
         case BUS_MATCH_LEAF:
 
                 if (bus) {
-                        if (node->leaf.callback->last_iteration == bus->iteration_counter)
-                                return 0;
+                        /* Don't run this match as long as the AddMatch() call is not complete yet.
+                         *
+                         * Don't run this match unless the 'after' counter has been reached.
+                         *
+                         * Don't run this match more than once per iteration */
+
+                        if (node->leaf.callback->install_slot ||
+                            m->read_counter <= node->leaf.callback->after ||
+                            node->leaf.callback->last_iteration == bus->iteration_counter)
+                                return bus_match_run(bus, node->next, m);
 
                         node->leaf.callback->last_iteration = bus->iteration_counter;
                 }

src/libsystemd/sd-bus/sd-bus.c

@@ -3327,7 +3327,7 @@ static int bus_add_match_full(
                                  * then make it floating. */
                                 r = sd_bus_slot_set_floating(s->match_callback.install_slot, true);
                         } else
-                                r = bus_add_match_internal(bus, s->match_callback.match_string);
+                                r = bus_add_match_internal(bus, s->match_callback.match_string, &s->match_callback.after);
                         if (r < 0)
                                 goto finish;