diff --git a/src/libsystemd/sd-bus/bus-message.c b/src/libsystemd/sd-bus/bus-message.c
index 0c3b292af8..700c57d362 100644
--- a/src/libsystemd/sd-bus/bus-message.c
+++ b/src/libsystemd/sd-bus/bus-message.c
@@ -118,7 +118,8 @@ static sd_bus_message* message_free(sd_bus_message *m) {
 
         message_reset_parts(m);
 
-        sd_bus_unref(m->bus);
+        /* Note that we don't unref m->bus here. That's already done by sd_bus_message_unref() as each user
+         * reference to the bus message also is considered a reference to the bus connection itself. */
 
         if (m->free_fds) {
                 close_many(m->fds, m->n_fds);
@@ -889,7 +890,80 @@ int bus_message_new_synthetic_error(
         return 0;
 }
 
-DEFINE_PUBLIC_TRIVIAL_REF_UNREF_FUNC(sd_bus_message, sd_bus_message, message_free);
+
+_public_ sd_bus_message* sd_bus_message_ref(sd_bus_message *m) {
+        if (!m)
+                return NULL;
+
+        /* We are fine if this message so far was either explicitly reffed or not reffed but queued into at
+         * least one bus connection object. */
+        assert(m->n_ref > 0 || m->n_queued > 0);
+
+        m->n_ref++;
+
+        /* Each user reference to a bus message shall also be considered a ref on the bus */
+        sd_bus_ref(m->bus);
+        return m;
+}
+
+_public_ sd_bus_message* sd_bus_message_unref(sd_bus_message *m) {
+        if (!m)
+                return NULL;
+
+        assert(m->n_ref > 0);
+
+        sd_bus_unref(m->bus); /* Each regular ref is also a ref on the bus connection. Let's hence drop it
+                               * here. Note we have to do this before decrementing our own n_ref here, since
+                               * otherwise, if this message is currently queued sd_bus_unref() might call
+                               * bus_message_unref_queued() for this which might then destroy the message
+                               * while we are still processing it. */
+        m->n_ref--;
+
+        if (m->n_ref > 0 || m->n_queued > 0)
+                return NULL;
+
+        /* Unset the bus field if neither the user has a reference nor this message is queued. We are careful
+         * to reset the field only after the last reference to the bus is dropped, after all we might keep
+         * multiple references to the bus, once for each reference kept on outselves. */
+        m->bus = NULL;
+
+        return message_free(m);
+}
+
+sd_bus_message* bus_message_ref_queued(sd_bus_message *m, sd_bus *bus) {
+        if (!m)
+                return NULL;
+
+        /* If this is a different bus than the message is associated with, then implicitly turn this into a
+         * regular reference. This means that you can create a memory leak by enqueuing a message generated
+         * on one bus onto another at the same time as enqueueing a message from the second one on the first,
+         * as we'll not detect the cyclic references there. */
+        if (bus != m->bus)
+                return sd_bus_message_ref(m);
+
+        assert(m->n_ref > 0 || m->n_queued > 0);
+        m->n_queued++;
+
+        return m;
+}
+
+sd_bus_message* bus_message_unref_queued(sd_bus_message *m, sd_bus *bus) {
+        if (!m)
+                return NULL;
+
+        if (bus != m->bus)
+                return sd_bus_message_unref(m);
+
+        assert(m->n_queued > 0);
+        m->n_queued--;
+
+        if (m->n_ref > 0 || m->n_queued > 0)
+                return NULL;
+
+        m->bus = NULL;
+
+        return message_free(m);
+}
 
 _public_ int sd_bus_message_get_type(sd_bus_message *m, uint8_t *type) {
         assert_return(m, -EINVAL);
diff --git a/src/libsystemd/sd-bus/bus-message.h b/src/libsystemd/sd-bus/bus-message.h
index 0115437d26..a7c4f81c4b 100644
--- a/src/libsystemd/sd-bus/bus-message.h
+++ b/src/libsystemd/sd-bus/bus-message.h
@@ -48,7 +48,16 @@ struct bus_body_part {
 };
 
 struct sd_bus_message {
-        unsigned n_ref;
+        /* Caveat: a message can be referenced in two different ways: the main (user-facing) way will also
+         * pin the bus connection object the message is associated with. The secondary way ("queued") is used
+         * when a message is in the read or write queues of the bus connection object, which will not pin the
+         * bus connection object. This is necessary so that we don't have to have a pair of cyclic references
+         * between a message that is queued and its connection: as soon as a message is only referenced by
+         * the connection (by means of being queued) and the connection itself has no other references it
+         * will be freed. */
+
+        unsigned n_ref;     /* Counter of references that pin the connection */
+        unsigned n_queued;  /* Counter of references that do not pin the connection */
 
         sd_bus *bus;
 
@@ -211,3 +220,6 @@ int bus_message_remarshal(sd_bus *bus, sd_bus_message **m);
 
 void bus_message_set_sender_driver(sd_bus *bus, sd_bus_message *m);
 void bus_message_set_sender_local(sd_bus *bus, sd_bus_message *m);
+
+sd_bus_message* bus_message_ref_queued(sd_bus_message *m, sd_bus *bus);
+sd_bus_message* bus_message_unref_queued(sd_bus_message *m, sd_bus *bus);