Merge 550f38cb14 into 2e5b0412f9

Follow-up for 2b07a3211b.

Fixes the failure found in
https://autopkgtest.ubuntu.com/results/autopkgtest-noble-upstream-systemd-ci-systemd-ci/noble/amd64/s/systemd-upstream/20241115_182040_92382@/log.gz
. Relevant logs:
```
Nov 16 02:48:36 systemd-networkd[2706]: veth99: Reconfiguring with /run/systemd/network/25-dhcp-client-ipv6-only.network.
Nov 16 02:48:36 systemd-networkd[2706]: veth99: NDISC: Started IPv6 Router Solicitation client
Nov 16 02:48:36 systemd-networkd[2706]: veth99: IPv6 Router Discovery is configured and started.
Nov 16 02:48:36 systemd-networkd[2706]: veth99: NDISC: Sent Router Solicitation, next solicitation in 3s
Nov 16 02:48:36 systemd-networkd[2706]: veth99: NDISC: Received Router Advertisement from fe80::1034:56ff:fe78:9abd: flags=0xc0(managed, other), preference=medium, lifetime=30min
Nov 16 02:48:36 systemd-networkd[2706]: veth99: NDISC: Invoking callback for 'router' event.
Nov 16 02:48:36 systemd-networkd[2706]: veth99: link_check_ready(): dynamic addressing protocols are enabled but none of them finished yet.
Nov 16 02:48:36 systemd-networkd[2706]: veth99: DHCPv6 client: Starting in Solicit mode
Nov 16 02:48:36 systemd-networkd[2706]: veth99: DHCPv6 client: State changed: stopped -> solicitation
Nov 16 02:48:36 systemd-networkd[2706]: veth99: Acquiring DHCPv6 lease on NDisc request
Nov 16 02:48:36 systemd-networkd[2706]: veth99: DHCPv6 client: Sent Solicit
Nov 16 02:48:36 systemd-networkd[2706]: veth99: DHCPv6 client: Next retransmission in 1s
Nov 16 02:48:37 systemd-networkd[2706]: veth99: DHCPv6 client: Sent Solicit
Nov 16 02:48:37 systemd-networkd[2706]: veth99: DHCPv6 client: Next retransmission in 1s
Nov 16 02:48:39 systemd-networkd[2706]: veth99: NDISC: Received Neighbor Advertisement from fe80::1034:56ff:fe78:9abd: Router=yes, Solicited=yes, Override=no
Nov 16 02:48:39 systemd-networkd[2706]: veth99: NDISC: Invoking callback for 'neighbor' event.
Nov 16 02:48:39 systemd-networkd[2706]: veth99: DHCPv6 client: Processed Reply message
Nov 16 02:48:39 systemd-networkd[2706]: veth99: DHCPv6 client: T1 expires in 50s
Nov 16 02:48:39 systemd-networkd[2706]: veth99: DHCPv6 client: T2 expires in 55s
Nov 16 02:48:39 systemd-networkd[2706]: veth99: DHCPv6 client: Valid lifetime expires in 2min
Nov 16 02:48:39 systemd-networkd[2706]: veth99: DHCPv6 client: State changed: solicitation -> bound
Nov 16 02:48:39 systemd-networkd[2706]: veth99: DHCPv6 address 2600::15/128 (valid for 1min 59s, preferred for 1min 59s)
Nov 16 02:48:41 systemd-networkd[2706]: veth99: Received updated DHCPv6 address (configured): 2600::15/128 (valid for 1min 58s, preferred for 1min 58s), flags: no-prefixroute, scope: global
Nov 16 02:48:41 systemd-networkd[2706]: veth99: DHCPv6 addresses and routes set.
Nov 16 02:48:41 systemd-networkd[2706]: veth99: link_check_ready(): IPv4LL:no DHCPv4:no DHCPv6:yes DHCP-PD:no NDisc:no
Nov 16 02:48:41 systemd-networkd[2706]: veth99: State changed: configuring -> configured
```
The interface veth99 entered the configured state after 5 seconds, but
at the same time, the `wait_online()` in the test script considered the
test failed.
The function `wait_online()` first invokes
`systemd-networkd-wait-online` with `--timeout=20`, then check setup
states of interfaces with 5 seconds timeout. So, the failure suggests
that `systemd-networkd-wait-online` finishes immediately, as the state
file was not updated when it is invoked, and thus it handles the
interface veth99 already in the configured state.

src/core/dbus-job.c

@@ -247,7 +247,7 @@ void bus_job_send_change_signal(Job *j) {
                 job_add_to_gc_queue(j);
         }
 
-        r = bus_foreach_bus(j->manager, j->bus_track, j->sent_dbus_new_signal ? send_changed_signal : send_new_signal, j);
+        r = bus_foreach_bus_signal(j->manager, j->bus_track, j->sent_dbus_new_signal ? send_changed_signal : send_new_signal, j);
         if (r < 0)
                 log_debug_errno(r, "Failed to send job change signal for %u: %m", j->id);
 
@@ -308,7 +308,7 @@ void bus_job_send_removed_signal(Job *j) {
         /* Make sure that any change signal on the unit is reflected before we send out the change signal on the job */
         bus_unit_send_pending_change_signal(j->unit, true);
 
-        r = bus_foreach_bus(j->manager, j->bus_track, send_removed_signal, j);
+        r = bus_foreach_bus_signal(j->manager, j->bus_track, send_removed_signal, j);
         if (r < 0)
                 log_debug_errno(r, "Failed to send job remove signal for %u: %m", j->id);
 }

src/core/dbus-manager.c

@@ -1337,10 +1337,6 @@ static int method_subscribe(sd_bus_message *message, void *userdata, sd_bus_erro
                 return r;
 
         if (sd_bus_message_get_bus(message) == m->api_bus) {
-
-                /* Note that direct bus connection subscribe by
-                 * default, we only track peers on the API bus here */
-
                 if (!m->subscribed) {
                         r = sd_bus_track_new(sd_bus_message_get_bus(message), &m->subscribed, NULL, NULL);
                         if (r < 0)
@@ -1350,10 +1346,15 @@ static int method_subscribe(sd_bus_message *message, void *userdata, sd_bus_erro
                 r = sd_bus_track_add_sender(m->subscribed, message);
                 if (r < 0)
                         return r;
-                if (r == 0)
-                        return sd_bus_error_set(error, BUS_ERROR_ALREADY_SUBSCRIBED, "Client is already subscribed.");
+        } else {
+                r = set_ensure_put(&m->private_buses_subscribed, NULL, sd_bus_message_get_bus(message));
+                if (r < 0)
+                        return r;
         }
 
+        if (r == 0)
+                return sd_bus_error_set(error, BUS_ERROR_ALREADY_SUBSCRIBED, "Client is already subscribed.");
+
         return sd_bus_reply_method_return(message, NULL);
 }
 
@@ -1373,9 +1374,11 @@ static int method_unsubscribe(sd_bus_message *message, void *userdata, sd_bus_er
                 r = sd_bus_track_remove_sender(m->subscribed, message);
                 if (r < 0)
                         return r;
-                if (r == 0)
-                        return sd_bus_error_set(error, BUS_ERROR_NOT_SUBSCRIBED, "Client is not subscribed.");
-        }
+        } else
+                r = !!set_remove(m->private_buses_subscribed, sd_bus_message_get_bus(message));
+
+        if (r == 0)
+                return sd_bus_error_set(error, BUS_ERROR_NOT_SUBSCRIBED, "Client is not subscribed.");
 
         return sd_bus_reply_method_return(message, NULL);
 }
@@ -2339,7 +2342,7 @@ static void manager_unit_files_changed(Manager *m, const InstallChange *changes,
         /* See comments for this variable in manager.h */
         m->unit_file_state_outdated = true;
 
-        r = bus_foreach_bus(m, NULL, send_unit_files_changed, NULL);
+        r = bus_foreach_bus_signal(m, NULL, send_unit_files_changed, NULL);
         if (r < 0)
                 log_debug_errno(r, "Failed to send UnitFilesChanged signal, ignoring: %m");
 }
@@ -3706,7 +3709,7 @@ void bus_manager_send_finished(
 
         assert(m);
 
-        r = bus_foreach_bus(
+        r = bus_foreach_bus_signal(
                         m,
                         NULL,
                         send_finished,
@@ -3744,7 +3747,7 @@ void bus_manager_send_reloading(Manager *m, bool active) {
 
         assert(m);
 
-        r = bus_foreach_bus(m, NULL, send_reloading, INT_TO_PTR(active));
+        r = bus_foreach_bus_signal(m, NULL, send_reloading, INT_TO_PTR(active));
         if (r < 0)
                 log_debug_errno(r, "Failed to send reloading signal: %m");
 }
@@ -3763,7 +3766,7 @@ void bus_manager_send_change_signal(Manager *m) {
 
         assert(m);
 
-        r = bus_foreach_bus(m, NULL, send_changed_signal, NULL);
+        r = bus_foreach_bus_signal(m, NULL, send_changed_signal, NULL);
         if (r < 0)
                 log_debug_errno(r, "Failed to send manager change signal: %m");
 }

src/core/dbus-unit.c

@@ -1705,7 +1705,7 @@ void bus_unit_send_change_signal(Unit *u) {
         if (!u->id)
                 return;
 
-        r = bus_foreach_bus(u->manager, u->bus_track, u->sent_dbus_new_signal ? send_changed_signal : send_new_signal, u);
+        r = bus_foreach_bus_signal(u->manager, u->bus_track, u->sent_dbus_new_signal ? send_changed_signal : send_new_signal, u);
         if (r < 0)
                 log_unit_debug_errno(u, r, "Failed to send unit change signal for %s: %m", u->id);
 
@@ -1800,7 +1800,7 @@ void bus_unit_send_removed_signal(Unit *u) {
         if (!u->id)
                 return;
 
-        r = bus_foreach_bus(u->manager, u->bus_track, send_removed_signal, u);
+        r = bus_foreach_bus_signal(u->manager, u->bus_track, send_removed_signal, u);
         if (r < 0)
                 log_unit_debug_errno(u, r, "Failed to send unit remove signal for %s: %m", u->id);
 }

src/core/dbus.c

@@ -138,6 +138,9 @@ static int signal_disconnected(sd_bus_message *message, void *userdata, sd_bus_e
 
         if (set_remove(m->private_buses, bus)) {
                 log_debug("Got disconnect on private connection.");
+
+                /* Don't bother checking if the bus was subscribed; try to remove it opportunistically. */
+                set_remove(m->private_buses_subscribed, bus);
                 destroy_bus(m, &bus);
         }
 
@@ -761,6 +764,9 @@ static int bus_on_connection(sd_event_source *s, int fd, uint32_t revents, void
                 return 0;
         }
 
+        /* If this bus (i.e. object address) was subscribed previously let's drop it from that set. */
+        set_remove(m->private_buses_subscribed, bus);
+
         TAKE_PTR(bus);
 
         log_debug("Accepted new private connection.");
@@ -1044,6 +1050,8 @@ void bus_done_private(Manager *m) {
                 destroy_bus(m, &b);
 
         m->private_buses = set_free(m->private_buses);
+        m->private_buses_subscribed = set_free(m->private_buses_subscribed);
+
 
         m->private_listen_event_source = sd_event_source_disable_unref(m->private_listen_event_source);
         m->private_listen_fd = safe_close(m->private_listen_fd);
@@ -1101,33 +1109,33 @@ int bus_fdset_add_all(Manager *m, FDSet *fds) {
         return 0;
 }
 
-int bus_foreach_bus(
+int bus_foreach_bus_signal(
                 Manager *m,
                 sd_bus_track *subscribed2,
-                int (*send_message)(sd_bus *bus, void *userdata),
+                int (*send_signal)(sd_bus *bus, void *userdata),
                 void *userdata) {
 
         int r = 0;
 
         assert(m);
-        assert(send_message);
+        assert(send_signal);
 
         /* Send to all direct buses, unconditionally */
         sd_bus *b;
-        SET_FOREACH(b, m->private_buses) {
+        SET_FOREACH(b, m->private_buses_subscribed) {
 
                 /* Don't bother with enqueuing these messages to clients that haven't started yet */
                 if (sd_bus_is_ready(b) <= 0)
                         continue;
 
-                RET_GATHER(r, send_message(b, userdata));
+                RET_GATHER(r, send_signal(b, userdata));
         }
 
         /* Send to API bus, but only if somebody is subscribed */
         if (m->api_bus &&
             (sd_bus_track_count(m->subscribed) > 0 ||
              sd_bus_track_count(subscribed2) > 0))
-                RET_GATHER(r, send_message(m->api_bus, userdata));
+                RET_GATHER(r, send_signal(m->api_bus, userdata));
 
         return r;
 }

src/core/dbus.h

@@ -21,7 +21,7 @@ int bus_fdset_add_all(Manager *m, FDSet *fds);
 void bus_track_serialize(sd_bus_track *t, FILE *f, const char *prefix);
 int bus_track_coldplug(Manager *m, sd_bus_track **t, bool recursive, char **l);
 
-int bus_foreach_bus(Manager *m, sd_bus_track *subscribed2, int (*send_message)(sd_bus *bus, void *userdata), void *userdata);
+int bus_foreach_bus_signal(Manager *m, sd_bus_track *subscribed2, int (*send_signal)(sd_bus *bus, void *userdata), void *userdata);
 
 int bus_forward_agent_released(Manager *m, const char *path);
 

src/core/manager.h

@@ -332,6 +332,8 @@ struct Manager {
         /* Data specific to the D-Bus subsystem */
         sd_bus *api_bus, *system_bus;
         Set *private_buses;
+        /* Private buses on which client called Subscribe() method */
+        Set *private_buses_subscribed;
         int private_listen_fd;
         sd_event_source *private_listen_event_source;
 

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

@@ -327,6 +327,8 @@ struct sd_bus {
 
         /* zero means use value specified by $SYSTEMD_BUS_TIMEOUT= environment variable or built-in default */
         usec_t method_call_timeout;
+
+        bool queue_toggle;
 };
 
 /* For method calls we timeout at 25s, like in the D-Bus reference implementation */

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

@@ -3064,9 +3064,14 @@ static int process_running(sd_bus *bus, sd_bus_message **ret) {
         if (r != 0)
                 goto null_message;
 
-        r = dispatch_wqueue(bus);
-        if (r != 0)
-                goto null_message;
+        /* Toggle which queue we process preferably in this iteration, the rq or the wq */
+        bus->queue_toggle = !bus->queue_toggle;
+
+        if (bus->queue_toggle) { /* wqueue preferably */
+                r = dispatch_wqueue(bus);
+                if (r != 0)
+                        goto null_message;
+        }
 
         r = dispatch_track(bus);
         if (r != 0)
@@ -3075,6 +3080,12 @@ static int process_running(sd_bus *bus, sd_bus_message **ret) {
         r = dispatch_rqueue(bus, &m);
         if (r < 0)
                 return r;
+        if (r == 0 && !bus->queue_toggle) {
+                /* Nothing happened on the rq, and we didn't check the wq in this iteration, hence do that now. */
+                r = dispatch_wqueue(bus);
+                goto null_message;
+        }
+
         if (!m)
                 goto null_message;
 

src/network/networkd-link.c

@@ -1443,6 +1443,7 @@ int link_reconfigure_impl(Link *link, LinkReconfigurationFlag flags) {
 }
 
 typedef struct LinkReconfigurationData {
+        Manager *manager;
         Link *link;
         LinkReconfigurationFlag flags;
         sd_bus_message *message;
@@ -1473,6 +1474,12 @@ static void link_reconfiguration_data_destroy_callback(LinkReconfigurationData *
                 }
 
                 if (!data->counter || *data->counter <= 0) {
+                        /* Update the state files before replying the bus method. Otherwise,
+                         * systemd-networkd-wait-online following networkctl reload/reconfigure may read an
+                         * outdated state file and wrongly handle an interface is already in the configured
+                         * state. */
+                        (void) manager_clean_all(data->manager);
+
                         r = sd_bus_reply_method_return(data->message, NULL);
                         if (r < 0)
                                 log_warning_errno(r, "Failed to reply for DBus method, ignoring: %m");
@@ -1521,6 +1528,7 @@ int link_reconfigure_full(Link *link, LinkReconfigurationFlag flags, sd_bus_mess
         }
 
         *data = (LinkReconfigurationData) {
+                .manager = link->manager,
                 .link = link_ref(link),
                 .flags = flags,
                 .message = sd_bus_message_ref(message), /* message may be NULL, but _ref() works fine. */

src/shared/bus-wait-for-jobs.c

@@ -112,6 +112,19 @@ int bus_wait_for_jobs_new(sd_bus *bus, BusWaitForJobs **ret) {
         if (r < 0)
                 return r;
 
+        r = sd_bus_call_method_async(
+                        bus,
+                        /* slot= */ NULL,
+                        "org.freedesktop.systemd1",
+                        "/org/freedesktop/systemd1",
+                        "org.freedesktop.systemd1.Manager",
+                        "Subscribe",
+                        /* callback= */ NULL,
+                        /* userdata= */ NULL,
+                        /* types= */ NULL);
+        if (r < 0)
+                return r;
+
         *ret = TAKE_PTR(d);
 
         return 0;

test/test-network/systemd-networkd-tests.py

@@ -6406,11 +6406,11 @@ class NetworkdRATests(unittest.TestCase, Utilities):
 
         for i in [100, 200, 300, 512, 1024, 2048]:
             if i not in [metric_1, metric_2]:
-                self.assertNotIn(f'{i}', output)
+                self.assertNotIn(f'metric {i} ', output)
 
         for i in ['low', 'medium', 'high']:
             if i not in [preference_1, preference_2]:
-                self.assertNotIn(f'{i}', output)
+                self.assertNotIn(f'pref {i}', output)
 
     def test_router_preference(self):
         copy_network_unit('25-veth-client.netdev',