Merge 015de427ed into b1236ce38b

On musl, O_ACCMODE is defined as (03|O_SEARCH), unlike glibc which
defines it as (O_RDONLY|O_WRONLY|O_RDWR). Additionally, O_SEARCH is
simply defined as O_PATH.

This causes problems for systemd on musl, as it changes the
behaviour of open_mkdir_at_full() to return -EINVAL if O_PATH is
included in flags due to the fact that O_ACCMODE includes O_SEARCH
(i.e. O_PATH). Consequently, this makes the test-fs-util test fail.

Upstream musl seems content with this behaviour and doesn't seem
interested in matching glibc's behaviour due to that defining it this
way allows for O_SEARCH to match POSIX better by allowing it to open
directories where read permission is missing. Apparently musl does some
emulation in other places to make this work more consistently as well.

Initially I took the approach of working around this by redefining
O_SEARCH as O_RDONLY if O_SEARCH == O_PATH. This fixes the test and is
the approach taken by both XZ[1] and Gzip[2][3], but was not taken as
redefining system headers potentially could be problematic.

Instead, introduce O_ACCMODE_STRICT which just is a copy of glibc's
O_ACCMODE and use it everywhere. This way we don't have to deal with
unusual definitions of O_ACCMODE from C standard libraries other than
glibc.

 [1]: https://git.tukaani.org/?p=xz.git;a=blob;f=src/xz/file_io.c;h=8c83269b13fa31284f7ea5f3627a1dfbce7d6e14;hb=HEAD#l72
 [2]: https://git.savannah.gnu.org/cgit/gnulib.git/tree/lib/fcntl.in.h
      (lines 380 and 396, commit d7f551b30f3f2a0fa57c1b10c12f4eea41a9b89e)
 [3]: https://lists.gnu.org/archive/html/bug-gzip/2025-01/msg00000.html

.github/workflows/coverage.yml

@@ -25,7 +25,7 @@ jobs:
 
     steps:
       - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683
-      - uses: systemd/mkosi@7e4ec15aee6b98300b2ee14265bc647a716a9f8a
+      - uses: systemd/mkosi@dbb4020beee2cdf250f93a425794f1cf8b0fe693
 
       # Freeing up disk space with rm -rf can take multiple minutes. Since we don't need the extra free space
       # immediately, we remove the files in the background. However, we first move them to a different location
@@ -90,7 +90,6 @@ jobs:
           sudo mkosi sandbox -- \
             meson setup \
             --buildtype=debugoptimized \
-            -Dintegration-tests=true \
             build
 
       - name: Build image
@@ -120,7 +119,8 @@ jobs:
             meson test \
             -C build \
             --no-rebuild \
-            --suite integration-tests \
+            --setup=integration \
+            --suite=integration-tests \
             --print-errorlogs \
             --no-stdsplit \
             --num-processes "$(($(nproc) - 1))" \

.github/workflows/mkosi.yml

@@ -120,7 +120,7 @@ jobs:
 
     steps:
       - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683
-      - uses: systemd/mkosi@7e4ec15aee6b98300b2ee14265bc647a716a9f8a
+      - uses: systemd/mkosi@dbb4020beee2cdf250f93a425794f1cf8b0fe693
 
       # Freeing up disk space with rm -rf can take multiple minutes. Since we don't need the extra free space
       # immediately, we remove the files in the background. However, we first move them to a different location
@@ -197,7 +197,6 @@ jobs:
           sudo mkosi sandbox -- \
             meson setup \
             --buildtype=debugoptimized \
-            -Dintegration-tests=true \
             -Dbpf-framework=disabled \
             build
 
@@ -233,7 +232,8 @@ jobs:
             meson test \
             -C build \
             --no-rebuild \
-            --suite integration-tests \
+            --setup=integration \
+            --suite=integration-tests \
             --print-errorlogs \
             --no-stdsplit \
             --num-processes "$(($(nproc) - 1))" \

TODO

@@ -135,17 +135,18 @@ Features:
 
 * run0: maybe enable utmp for run0 sessions, so that they are easily visible.
 
+* maybe beef up sd-event: optionally, allow sd-event to query the timestamp of
+  next pending datagram inside a SOCK_DGRAM IO fd, and order event source
+  dispatching by that. Enable this on the native + syslog sockets in journald,
+  so that we add correct ordering between the two. Use MSG_PEEK + SCM_TIMESTAMP
+  for this.
+
 * maybe replace nss-machines with logic in networkd that registers records with
   systemd-resolved, based on DHCP leases, so that we gain compat with VMs.
   Implementation idea: encode in an ifaltname the intended local name to expose this
   under and then parse that out and map it to the combined A/AAAA of all handed
   out leases.
 
-* make journalctl something we can invoke like the askpw/polkit agents, and
-  then spawn it from "systemctl start" and similar with a precise filter on the
-  unit. Make sure that it syncs on the journal when done so that we show
-  "complete" logs. Make this easily reachable via a new "-v" switch or so.
-
 * bsod: add target "bsod.target" or so, which invokes systemd-bsod.target and
   waits and then reboots. Then use OnFailure=bsod.target from various jobs that
   should result in system reboots, such as TPM tamper detection cases.
@@ -2440,7 +2441,6 @@ Features:
                 journalctl /usr/bin/X11 --invocation=-1
   - systemctl: change 'status' to show logs for the last invocation, not a fixed
     number of lines
-  - systemctl: expand --wait to show logs for the invocation with a new switch
   - improve journalctl performance by loading journal files
     lazily. Encode just enough information in the file name, so that we
     do not have to open it to know that it is not interesting for us, for
@@ -2710,11 +2710,6 @@ Features:
   ensure deterministic behaviour if two unit files conflict (like DMs
   do, for example)
 
-* add "systemctl start -v foobar.service" that shows logs of a service
-  while the start command runs. This is non-trivial to do without
-  races though, since we should flush out all journal messages before
-  returning from the "systemctl stop".
-
 * systemctl: if some operation fails, show log output?
 
 * Add a new verb "systemctl top"

man/journalctl.xml

@@ -792,6 +792,20 @@
         --"), any warning messages regarding inaccessible system journals when run as a normal
         user.</para></listitem>
       </varlistentry>
+
+      <varlistentry>
+        <term><option>--synchronize-on-exit=</option></term>
+
+        <listitem><para>Takes a boolean argument. If true and operating in <option>--follow</option> mode, a
+        journal synchronization request (equivalent to <command>journalctl --sync</command>) is issued when
+        <constant>SIGTERM</constant>/<constant>SIGINT</constant> is received, and log output continues until
+        this request completes. This is useful for synchronizing journal log output to the runtime of
+        services or external events, ensuring that any log data enqueued to the logging subsystem by
+        the time <constant>SIGTERM</constant>/<constant>SIGINT</constant> is issued is guaranteed to be
+        processed and displayed by the time log output ends. Defaults to false.</para>
+
+        <xi:include href="version-info.xml" xpointer="v258"/></listitem>
+      </varlistentry>
     </variablelist>
   </refsect1>
 

man/systemctl.xml

@@ -2349,6 +2349,17 @@ Jan 12 10:46:45 example.com bluetoothd[8900]: gatt-time-server: Input/output err
         </listitem>
       </varlistentry>
 
+      <varlistentry>
+        <term><option>-v</option></term>
+        <term><option>--verbose</option></term>
+
+        <listitem>
+          <para>Display unit log output while executing unit operations.</para>
+
+          <xi:include href="version-info.xml" xpointer="v258"/>
+        </listitem>
+      </varlistentry>
+
       <varlistentry>
         <term><option>--no-warn</option></term>
 

man/systemd-run.xml

@@ -394,6 +394,17 @@
         <xi:include href="version-info.xml" xpointer="v219"/></listitem>
       </varlistentry>
 
+      <varlistentry>
+        <term><option>-v</option></term>
+        <term><option>--verbose</option></term>
+
+        <listitem>
+          <para>Display unit log output while running.</para>
+
+          <xi:include href="version-info.xml" xpointer="v258"/>
+        </listitem>
+      </varlistentry>
+
       <varlistentry>
         <term><option>--on-active=</option></term>
         <term><option>--on-boot=</option></term>

meson.build

@@ -13,6 +13,12 @@ project('systemd', 'c',
         meson_version : '>= 0.62.0',
 )
 
+add_test_setup(
+        'default',
+        exclude_suites : ['integration-tests'],
+        is_default : true,
+)
+
 project_major_version = meson.project_version().split('.')[0].split('~')[0]
 if meson.project_version().contains('.')
         project_minor_version = meson.project_version().split('.')[-1].split('~')[0]
@@ -339,7 +345,6 @@ meson_build_sh = find_program('tools/meson-build.sh')
 want_tests = get_option('tests')
 want_slow_tests = want_tests != 'false' and get_option('slow-tests')
 want_fuzz_tests = want_tests != 'false' and get_option('fuzz-tests')
-want_integration_tests = want_tests != 'false' and get_option('integration-tests')
 install_tests = want_tests != 'false' and get_option('install-tests')
 
 if add_languages('cpp', native : false, required : fuzzer_build)
@@ -2661,10 +2666,6 @@ endif
 #####################################################################
 
 mkosi = find_program('mkosi', required : false)
-if want_integration_tests and not mkosi.found()
-        error('Could not find mkosi which is required to run the integration tests')
-endif
-
 mkosi_depends = public_programs
 
 foreach executable : ['systemd-journal-remote', 'systemd-sbsign', 'systemd-keyutil']

meson_options.txt

@@ -509,7 +509,7 @@ option('install-tests', type : 'boolean', value : false,
        description : 'install test executables')
 option('log-message-verification', type : 'feature', deprecated : { 'true' : 'enabled', 'false' : 'disabled' },
        description : 'do fake printf() calls to verify format strings')
-option('integration-tests', type : 'boolean', value : false,
+option('integration-tests', type : 'boolean', value : false, deprecated : true,
        description : 'run the integration tests')
 
 option('ok-color', type : 'combo',

mkosi/mkosi.conf

@@ -1,7 +1,7 @@
 # SPDX-License-Identifier: LGPL-2.1-or-later
 
 [Config]
-MinimumVersion=commit:7e4ec15aee6b98300b2ee14265bc647a716a9f8a
+MinimumVersion=commit:dbb4020beee2cdf250f93a425794f1cf8b0fe693
 Dependencies=
         exitrd
         initrd
@@ -78,8 +78,7 @@ KernelCommandLine=
         oops=panic
         panic=-1
         softlockup_panic=1
-        # Disabled due to BTRFS issue, waiting for the fix to become available
-        panic_on_warn=0
+        panic_on_warn=1
         psi=1
         mitigations=off
 

src/basic/compress.c

@@ -491,11 +491,10 @@ int decompress_blob_zstd(
         };
 
         size_t k = sym_ZSTD_decompressStream(dctx, &output, &input);
-        if (sym_ZSTD_isError(k)) {
-                log_debug("ZSTD decoder failed: %s", sym_ZSTD_getErrorName(k));
-                return zstd_ret_to_errno(k);
-        }
-        assert(output.pos >= size);
+        if (sym_ZSTD_isError(k))
+                return log_debug_errno(zstd_ret_to_errno(k), "ZSTD decoder failed: %s", sym_ZSTD_getErrorName(k));
+        if (output.pos < size)
+                return log_debug_errno(SYNTHETIC_ERRNO(EBADMSG), "ZSTD decoded less data than indicated, probably corrupted stream.");
 
         *dst_size = size;
         return 0;

src/basic/fd-util.c

@@ -1001,13 +1001,13 @@ int fd_verify_safe_flags_full(int fd, int extra_flags) {
         if (flags < 0)
                 return -errno;
 
-        unexpected_flags = flags & ~(O_ACCMODE|O_NOFOLLOW|RAW_O_LARGEFILE|extra_flags);
+        unexpected_flags = flags & ~(O_ACCMODE_STRICT|O_NOFOLLOW|RAW_O_LARGEFILE|extra_flags);
         if (unexpected_flags != 0)
                 return log_debug_errno(SYNTHETIC_ERRNO(EREMOTEIO),
                                        "Unexpected flags set for extrinsic fd: 0%o",
                                        (unsigned) unexpected_flags);
 
-        return flags & (O_ACCMODE | extra_flags); /* return the flags variable, but remove the noise */
+        return flags & (O_ACCMODE_STRICT | extra_flags); /* return the flags variable, but remove the noise */
 }
 
 int read_nr_open(void) {
@@ -1132,7 +1132,7 @@ int fds_are_same_mount(int fd1, int fd2) {
 }
 
 const char* accmode_to_string(int flags) {
-        switch (flags & O_ACCMODE) {
+        switch (flags & O_ACCMODE_STRICT) {
         case O_RDONLY:
                 return "ro";
         case O_WRONLY:

src/basic/fs-util.c

@@ -1036,7 +1036,7 @@ int open_mkdir_at_full(int dirfd, const char *path, int flags, XOpenFlags xopen_
 
         if (flags & ~(O_RDONLY|O_CLOEXEC|O_DIRECTORY|O_EXCL|O_NOATIME|O_NOFOLLOW|O_PATH))
                 return -EINVAL;
-        if ((flags & O_ACCMODE) != O_RDONLY)
+        if ((flags & O_ACCMODE_STRICT) != O_RDONLY)
                 return -EINVAL;
 
         /* Note that O_DIRECTORY|O_NOFOLLOW is implied, but we allow specifying it anyway. The following

src/basic/missing_fcntl.h

@@ -43,3 +43,9 @@
 #ifndef AT_HANDLE_FID
 #define AT_HANDLE_FID AT_REMOVEDIR
 #endif
+
+/* On musl, O_ACCMODE is defined as (03|O_SEARCH), unlike glibc which defines it as
+ * (O_RDONLY|O_WRONLY|O_RDWR). Additionally, O_SEARCH is simply defined as O_PATH. This changes the behaviour
+ * of O_ACCMODE in certain situations, which we don't want. This definition is copied from glibc and works
+ * around the problems with musl's definition. */
+#define O_ACCMODE_STRICT (O_RDONLY|O_WRONLY|O_RDWR)

src/basic/socket-util.c

@@ -1464,7 +1464,6 @@ int socket_autobind(int fd, char **ret_name) {
          * "autobind" feature, but uses 64-bit random number internally. */
 
         assert(fd >= 0);
-        assert(ret_name);
 
         random = random_u64();
 
@@ -1481,7 +1480,8 @@ int socket_autobind(int fd, char **ret_name) {
         if (bind(fd, &sa.sa, r) < 0)
                 return -errno;
 
-        *ret_name = TAKE_PTR(name);
+        if (ret_name)
+                *ret_name = TAKE_PTR(name);
         return 0;
 }
 

src/core/exec-invoke.c

@@ -267,7 +267,7 @@ static int acquire_path(const char *path, int flags, mode_t mode) {
 
         assert(path);
 
-        if (IN_SET(flags & O_ACCMODE, O_WRONLY, O_RDWR))
+        if (IN_SET(flags & O_ACCMODE_STRICT, O_WRONLY, O_RDWR))
                 flags |= O_CREAT;
 
         fd = open(path, flags|O_NOCTTY, mode);
@@ -291,9 +291,9 @@ static int acquire_path(const char *path, int flags, mode_t mode) {
         if (r < 0)
                 return r;
 
-        if ((flags & O_ACCMODE) == O_RDONLY)
+        if ((flags & O_ACCMODE_STRICT) == O_RDONLY)
                 r = shutdown(fd, SHUT_WR);
-        else if ((flags & O_ACCMODE) == O_WRONLY)
+        else if ((flags & O_ACCMODE_STRICT) == O_WRONLY)
                 r = shutdown(fd, SHUT_RD);
         else
                 r = 0;

src/journal-remote/journal-remote-main.c

@@ -97,10 +97,11 @@ static MHDDaemonWrapper* MHDDaemonWrapper_free(MHDDaemonWrapper *d) {
         if (!d)
                 return NULL;
 
+        d->io_event = sd_event_source_unref(d->io_event);
+        d->timer_event = sd_event_source_unref(d->timer_event);
+
         if (d->daemon)
                 MHD_stop_daemon(d->daemon);
-        sd_event_source_unref(d->io_event);
-        sd_event_source_unref(d->timer_event);
 
         return mfree(d);
 }

src/journal-remote/journal-upload-journal.c

@@ -28,6 +28,11 @@ static ssize_t write_entry(char *buf, size_t size, Uploader *u) {
                         u->current_cursor = mfree(u->current_cursor);
 
                         r = sd_journal_get_cursor(u->journal, &u->current_cursor);
+                        if (r == -EBADMSG) {
+                                log_debug("Encountered bad or partially written entry while acquiring cursor, leaving.");
+                                u->entry_state = ENTRY_OUTRO;
+                                continue;
+                        }
                         if (r < 0)
                                 return log_error_errno(r, "Failed to get cursor: %m");
 
@@ -53,6 +58,11 @@ static ssize_t write_entry(char *buf, size_t size, Uploader *u) {
                         usec_t realtime;
 
                         r = sd_journal_get_realtime_usec(u->journal, &realtime);
+                        if (r == -EBADMSG) {
+                                log_debug("Encountered bad or partially written realtime timestamp, leaving.");
+                                u->entry_state = ENTRY_OUTRO;
+                                continue;
+                        }
                         if (r < 0)
                                 return log_error_errno(r, "Failed to get realtime timestamp: %m");
 
@@ -79,6 +89,11 @@ static ssize_t write_entry(char *buf, size_t size, Uploader *u) {
                         sd_id128_t boot_id;
 
                         r = sd_journal_get_monotonic_usec(u->journal, &monotonic, &boot_id);
+                        if (r == -EBADMSG) {
+                                log_debug("Encountered bad or partially written monotonic timestamp, leaving.");
+                                u->entry_state = ENTRY_OUTRO;
+                                continue;
+                        }
                         if (r < 0)
                                 return log_error_errno(r, "Failed to get monotonic timestamp: %m");
 
@@ -103,7 +118,12 @@ static ssize_t write_entry(char *buf, size_t size, Uploader *u) {
                 case ENTRY_BOOT_ID: {
                         sd_id128_t boot_id;
 
-                        r = sd_journal_get_monotonic_usec(u->journal, NULL, &boot_id);
+                        r = sd_journal_get_monotonic_usec(u->journal, /* ret_monotonic= */ NULL, &boot_id);
+                        if (r == -EBADMSG) {
+                                log_debug("Encountered bad or partially written boot ID, leaving.");
+                                u->entry_state = ENTRY_OUTRO;
+                                continue;
+                        }
                         if (r < 0)
                                 return log_error_errno(r, "Failed to get monotonic timestamp: %m");
 
@@ -131,9 +151,14 @@ static ssize_t write_entry(char *buf, size_t size, Uploader *u) {
                         r = sd_journal_enumerate_data(u->journal,
                                                       &u->field_data,
                                                       &u->field_length);
+                        if (r == -EBADMSG) {
+                                log_debug("Encountered bad or partially written data field, leaving.");
+                                u->entry_state = ENTRY_OUTRO;
+                                continue;
+                        }
                         if (r < 0)
                                 return log_error_errno(r, "Failed to move to next field in entry: %m");
-                        else if (r == 0) {
+                        if (r == 0) {
                                 u->entry_state = ENTRY_OUTRO;
                                 continue;
                         }
@@ -171,10 +196,10 @@ static ssize_t write_entry(char *buf, size_t size, Uploader *u) {
                                 pos += tocopy + 1;
                                 u->entry_state = ENTRY_NEW_FIELD;
                                 continue;
-                        } else {
-                                u->field_pos += tocopy;
-                                return size;
                         }
+
+                        u->field_pos += tocopy;
+                        return size;
                 }
 
                 case ENTRY_BINARY_FIELD_START: {
@@ -182,9 +207,11 @@ static ssize_t write_entry(char *buf, size_t size, Uploader *u) {
                         size_t len;
 
                         c = memchr(u->field_data, '=', u->field_length);
-                        if (!c || c == u->field_data)
-                                return log_error_errno(SYNTHETIC_ERRNO(EINVAL),
-                                                       "Invalid field.");
+                        if (!c || c == u->field_data) {
+                                log_debug("Encountered field without '='. Assuming field is still being written, leaving.");
+                                u->entry_state = ENTRY_OUTRO;
+                                continue;
+                        }
 
                         len = c - (const char*)u->field_data;
 
@@ -198,8 +225,9 @@ static ssize_t write_entry(char *buf, size_t size, Uploader *u) {
 
                         u->field_pos = len + 1;
                         u->entry_state++;
-                }
+
                         _fallthrough_;
+                }
                 case ENTRY_BINARY_FIELD_SIZE: {
                         uint64_t le64;
 
@@ -274,10 +302,7 @@ static size_t journal_input_callback(void *buf, size_t size, size_t nmemb, void
         while (j && filled < size * nmemb) {
                 if (u->entry_state == ENTRY_DONE) {
                         r = sd_journal_next(j);
-                        if (r < 0) {
-                                log_error_errno(r, "Failed to move to next entry in journal: %m");
-                                return CURL_READFUNC_ABORT;
-                        } else if (r == 0) {
+                        if (r == 0) {
                                 if (u->input_event)
                                         log_debug("No more entries, waiting for journal.");
                                 else {
@@ -286,10 +311,23 @@ static size_t journal_input_callback(void *buf, size_t size, size_t nmemb, void
                                 }
 
                                 u->uploading = false;
-
                                 break;
                         }
+                        if (r == -EBADMSG) {
+                                if (u->input_event)
+                                        log_debug("Read bad or partially written entry, waiting for journal.");
+                                else {
+                                        log_info("Read bad or partially written entry, waiting for journal.");
+                                        close_journal_input(u);
+                                }
 
+                                u->uploading = false;
+                                break;
+                        }
+                        if (r < 0) {
+                                log_error_errno(r, "Failed to move to next entry in journal: %m");
+                                return CURL_READFUNC_ABORT;
+                        }
                         u->entry_state = ENTRY_CURSOR;
                 }
 

src/journal/journalctl-show.c

@@ -2,7 +2,9 @@
 
 #include <unistd.h>
 
+#include "sd-daemon.h"
 #include "sd-event.h"
+#include "sd-varlink.h"
 
 #include "ansi-color.h"
 #include "fileio.h"
@@ -10,6 +12,7 @@
 #include "journalctl-filter.h"
 #include "journalctl-show.h"
 #include "journalctl-util.h"
+#include "journalctl-varlink.h"
 #include "logs-show.h"
 #include "terminal-util.h"
 
@@ -25,11 +28,15 @@ typedef struct Context {
         sd_id128_t previous_boot_id;
         sd_id128_t previous_boot_id_output;
         dual_timestamp previous_ts_output;
+        sd_event *event;
+        sd_varlink *synchronize_varlink;
 } Context;
 
 static void context_done(Context *c) {
         assert(c);
 
+        c->synchronize_varlink = sd_varlink_flush_close_unref(c->synchronize_varlink);
+        c->event = sd_event_unref(c->event);
         sd_journal_close(c->journal);
 }
 
@@ -268,15 +275,14 @@ static int show(Context *c) {
         return n_shown;
 }
 
-static int show_and_fflush(Context *c, sd_event_source *s) {
+static int show_and_fflush(Context *c) {
         int r;
 
         assert(c);
-        assert(s);
 
         r = show(c);
         if (r < 0)
-                return sd_event_exit(sd_event_source_get_event(s), r);
+                return sd_event_exit(c->event, r);
 
         fflush(stdout);
         return 0;
@@ -291,39 +297,136 @@ static int on_journal_event(sd_event_source *s, int fd, uint32_t revents, void *
         r = sd_journal_process(c->journal);
         if (r < 0) {
                 log_error_errno(r, "Failed to process journal events: %m");
-                return sd_event_exit(sd_event_source_get_event(s), r);
+                return sd_event_exit(c->event, r);
         }
 
-        return show_and_fflush(c, s);
+        return show_and_fflush(c);
 }
 
 static int on_first_event(sd_event_source *s, void *userdata) {
-        return show_and_fflush(userdata, s);
+        Context *c = ASSERT_PTR(userdata);
+        int r;
+
+        assert(s);
+
+        r = show_and_fflush(c);
+        if (r < 0)
+                return r;
+
+        if (arg_follow && !arg_reverse && !arg_cursor && !arg_after_cursor && !arg_cursor_file && !arg_since_set) {
+                r = sd_journal_get_cursor(c->journal, /* ret_cursor= */ NULL);
+                if (r == -EADDRNOTAVAIL) {
+                        /* If we shall operate in --follow mode, and we are unable to get a cursor after
+                         * doing our first round of output, then this means there was no data to show
+                         * whatsoever, and we hence have no stable position on any line at all. This means,
+                         * when we get notified about changes, we shouldn't try to position the cursor at the
+                         * end of the logs anymore, but at the beginning, since anythng showing up from now
+                         * that matches our filters is good now. Hence, simply disable the effect of --lines=
+                         * now. */
+
+                        r = sd_journal_seek_head(c->journal);
+                        if (r < 0)
+                                return log_error_errno(r, "Failed to seek to head: %m");
+
+                        c->need_seek = true;
+
+                } else if (r < 0)
+                        return log_error_errno(r, "Failed to get cursor: %m");
+        }
+
+        (void) sd_notify(/* unset_environment= */ false, "READY=1");
+        return 0;
+}
+
+static int on_synchronize_reply(
+                sd_varlink *vl,
+                sd_json_variant *parameters,
+                const char *error_id,
+                sd_varlink_reply_flags_t flags,
+                void *userdata) {
+
+        Context *c = ASSERT_PTR(userdata);
+        int r;
+
+        assert(vl);
+
+        if (error_id)
+                log_warning("Failed to synchronize on Journal, ignoring: %s", error_id);
+
+        r = show_and_fflush(c);
+        if (r < 0)
+                return r;
+
+        return sd_event_exit(c->event, EXIT_SUCCESS);
 }
 
 static int on_signal(sd_event_source *s, const struct signalfd_siginfo *si, void *userdata) {
+        _cleanup_(sd_varlink_flush_close_unrefp) sd_varlink *vl = NULL;
+        Context *c = ASSERT_PTR(userdata);
+        int r;
+
         assert(s);
         assert(si);
         assert(IN_SET(si->ssi_signo, SIGTERM, SIGINT));
 
-        return sd_event_exit(sd_event_source_get_event(s), si->ssi_signo);
+        if (!arg_synchronize_on_exit)
+                goto finish;
+
+        if (c->synchronize_varlink) /* Already pending? Shortcut */
+                return 0;
+
+        r = varlink_connect_journal(&vl);
+        if (r < 0) {
+                log_error_errno(r, "Failed to connect to Journal Varlink IPC interface, ignoring: %m");
+                goto finish;
+        }
+
+        /* Set a low priority on the idle event handler, so that we show any log messages first */
+        r = sd_varlink_attach_event(vl, c->event, SD_EVENT_PRIORITY_IDLE);
+        if (r < 0) {
+                log_warning_errno(r, "Failed to attach Varlink connectio to event loop: %m");
+                goto finish;
+        }
+
+        r = sd_varlink_bind_reply(vl, on_synchronize_reply);
+        if (r < 0) {
+                log_warning_errno(r, "Failed to bind synchronization reply: %m");
+                goto finish;
+        }
+
+        (void) sd_varlink_set_userdata(vl, c);
+
+        r = sd_varlink_invokebo(
+                        vl,
+                        "io.systemd.Journal.Synchronize",
+                        SD_JSON_BUILD_PAIR_BOOLEAN("offline", false));
+        if (r < 0) {
+                log_warning_errno(r, "Failed to issue synchronization request: %m");
+                goto finish;
+        }
+
+        c->synchronize_varlink = TAKE_PTR(vl);
+        return 0;
+
+finish:
+        return sd_event_exit(c->event, si->ssi_signo);
 }
 
-static int setup_event(Context *c, int fd, sd_event **ret) {
-        _cleanup_(sd_event_unrefp) sd_event *e = NULL;
+static int setup_event(Context *c, int fd) {
         int r;
 
         assert(arg_follow);
         assert(c);
         assert(fd >= 0);
-        assert(ret);
+        assert(!c->event);
 
+        _cleanup_(sd_event_unrefp) sd_event *e = NULL;
         r = sd_event_default(&e);
         if (r < 0)
                 return log_error_errno(r, "Failed to allocate sd_event object: %m");
 
-        (void) sd_event_add_signal(e, NULL, SIGTERM | SD_EVENT_SIGNAL_PROCMASK, on_signal, NULL);
-        (void) sd_event_add_signal(e, NULL, SIGINT | SD_EVENT_SIGNAL_PROCMASK, on_signal, NULL);
+        (void) sd_event_add_signal(e, /* ret_event_source= */ NULL, SIGTERM | SD_EVENT_SIGNAL_PROCMASK, on_signal, c);
+        (void) sd_event_add_signal(e, /* ret_event_source= */ NULL, SIGINT | SD_EVENT_SIGNAL_PROCMASK, on_signal, c);
 
         r = sd_event_add_io(e, NULL, fd, EPOLLIN, &on_journal_event, c);
         if (r < 0)
@@ -345,7 +448,7 @@ static int setup_event(Context *c, int fd, sd_event **ret) {
                         return log_error_errno(r, "Failed to add defer event source: %m");
         }
 
-        *ret = TAKE_PTR(e);
+        c->event = TAKE_PTR(e);
         return 0;
 }
 
@@ -433,16 +536,15 @@ int action_show(char **matches) {
         }
 
         if (arg_follow) {
-                _cleanup_(sd_event_unrefp) sd_event *e = NULL;
                 int sig;
 
                 assert(poll_fd >= 0);
 
-                r = setup_event(&c, poll_fd, &e);
+                r = setup_event(&c, poll_fd);
                 if (r < 0)
                         return r;
 
-                r = sd_event_loop(e);
+                r = sd_event_loop(c.event);
                 if (r < 0)
                         return r;
                 sig = r;
@@ -455,6 +557,8 @@ int action_show(char **matches) {
                 return sig;
         }
 
+        (void) sd_notify(/* unset_environment= */ false, "READY=1");
+
         r = show(&c);
         if (r < 0)
                 return r;

src/journal/journalctl-varlink.c

@@ -10,7 +10,7 @@
 #include "journalctl-varlink.h"
 #include "varlink-util.h"
 
-static int varlink_connect_journal(sd_varlink **ret) {
+int varlink_connect_journal(sd_varlink **ret) {
         _cleanup_(sd_varlink_flush_close_unrefp) sd_varlink *vl = NULL;
         const char *address;
         int r;

src/journal/journalctl-varlink.h

@@ -1,6 +1,10 @@
 /* SPDX-License-Identifier: LGPL-2.1-or-later */
 #pragma once
 
+#include "sd-varlink.h"
+
+int varlink_connect_journal(sd_varlink **ret);
+
 int action_flush_to_var(void);
 int action_relinquish_var(void);
 int action_rotate(void);

src/journal/journalctl.c

@@ -92,7 +92,8 @@ Set *arg_output_fields = NULL;
 char *arg_pattern = NULL;
 pcre2_code *arg_compiled_pattern = NULL;
 PatternCompileCase arg_case = PATTERN_COMPILE_CASE_AUTO;
-static ImagePolicy *arg_image_policy = NULL;
+ImagePolicy *arg_image_policy = NULL;
+bool arg_synchronize_on_exit = false;
 
 STATIC_DESTRUCTOR_REGISTER(arg_cursor, freep);
 STATIC_DESTRUCTOR_REGISTER(arg_cursor_file, freep);
@@ -267,6 +268,8 @@ static int help(void) {
                "     --no-tail               Show all lines, even in follow mode\n"
                "     --truncate-newline      Truncate entries by first newline character\n"
                "  -q --quiet                 Do not show info messages and privilege warning\n"
+               "     --synchronize-on-exit=BOOL\n"
+               "                             Wait for Journal synchronization before exiting\n"
                "\n%3$sPager Control Options:%4$s\n"
                "     --no-pager              Do not pipe output into a pager\n"
                "  -e --pager-end             Immediately jump to the end in the pager\n"
@@ -355,6 +358,7 @@ static int parse_argv(int argc, char *argv[]) {
                 ARG_OUTPUT_FIELDS,
                 ARG_NAMESPACE,
                 ARG_LIST_NAMESPACES,
+                ARG_SYNCHRONIZE_ON_EXIT,
         };
 
         static const struct option options[] = {
@@ -428,6 +432,7 @@ static int parse_argv(int argc, char *argv[]) {
                 { "output-fields",        required_argument, NULL, ARG_OUTPUT_FIELDS        },
                 { "namespace",            required_argument, NULL, ARG_NAMESPACE            },
                 { "list-namespaces",      no_argument,       NULL, ARG_LIST_NAMESPACES      },
+                { "synchronize-on-exit",  required_argument, NULL, ARG_SYNCHRONIZE_ON_EXIT  },
                 {}
         };
 
@@ -971,6 +976,13 @@ static int parse_argv(int argc, char *argv[]) {
 
                         break;
                 }
+                case ARG_SYNCHRONIZE_ON_EXIT:
+                        r = parse_boolean_argument("--synchronize-on-exit", optarg, &arg_synchronize_on_exit);
+                        if (r < 0)
+                                return r;
+
+                        break;
+
                 case '?':
                         return -EINVAL;
 

src/journal/journalctl.h

@@ -7,6 +7,7 @@
 #include "sd-id128.h"
 #include "sd-json.h"
 
+#include "image-policy.h"
 #include "output-mode.h"
 #include "pager.h"
 #include "pcre2-util.h"
@@ -97,6 +98,8 @@ extern Set *arg_output_fields;
 extern char *arg_pattern;
 extern pcre2_code *arg_compiled_pattern;
 extern PatternCompileCase arg_case;
+extern ImagePolicy *arg_image_policy;
+extern bool arg_synchronize_on_exit;
 
 static inline bool arg_lines_needs_seek_end(void) {
         return arg_lines >= 0 && !arg_lines_oldest;

src/journal/journald-server.c

@@ -1594,6 +1594,9 @@ int server_process_datagram(
                         log_ratelimit_warning(JOURNAL_LOG_RATELIMIT,
                                               "Got file descriptors via syslog socket. Ignoring.");
 
+                if (tv)
+                        s->syslog_timestamp = timeval_load(tv);
+
         } else if (fd == s->native_fd) {
                 if (n > 0 && n_fds == 0)
                         server_process_native_message(s, s->buffer, n, ucred, tv, label, label_len);
@@ -1603,6 +1606,9 @@ int server_process_datagram(
                         log_ratelimit_warning(JOURNAL_LOG_RATELIMIT,
                                               "Got too many file descriptors via native socket. Ignoring.");
 
+                if (tv)
+                        s->native_timestamp = timeval_load(tv);
+
         } else {
                 assert(fd == s->audit_fd);
 
@@ -1615,6 +1621,9 @@ int server_process_datagram(
 
         close_many(fds, n_fds);
 
+        if (tv)
+                sync_req_revalidate_by_timestamp(s);
+
         server_refresh_idle_timer(s);
         return 0;
 }
@@ -2754,6 +2763,11 @@ Server* server_free(Server *s) {
 
         mmap_cache_unref(s->mmap);
 
+        SyncReq *req;
+        while ((req = prioq_peek(s->sync_req_prioq)))
+                sync_req_free(req);
+        prioq_free(s->sync_req_prioq);
+
         return mfree(s);
 }
 

src/journal/journald-server.h

@@ -185,6 +185,15 @@ struct Server {
         ClientContext *pid1_context; /* the context of PID 1 */
 
         sd_varlink_server *varlink_server;
+
+        /* timestamp of most recently processed log messages from each source (CLOCK_REALTIME) */
+        usec_t native_timestamp, syslog_timestamp;
+
+        /* Pending synchronization requests, ordered by their timestamp */
+        Prioq *sync_req_prioq;
+
+        /* Pending synchronization requests with non-zero rqlen counter */
+        LIST_HEAD(SyncReq, sync_req_pending_rqlen);
 };
 
 #define SERVER_MACHINE_ID(s) ((s)->machine_id_field + STRLEN("_MACHINE_ID="))

src/journal/journald-stream.c

@@ -48,17 +48,6 @@
  * let's enforce a line length matching the maximum unit name length (255) */
 #define STDOUT_STREAM_SETUP_PROTOCOL_LINE_MAX (UNIT_NAME_MAX-1U)
 
-typedef enum StdoutStreamState {
-        STDOUT_STREAM_IDENTIFIER,
-        STDOUT_STREAM_UNIT_ID,
-        STDOUT_STREAM_PRIORITY,
-        STDOUT_STREAM_LEVEL_PREFIX,
-        STDOUT_STREAM_FORWARD_TO_SYSLOG,
-        STDOUT_STREAM_FORWARD_TO_KMSG,
-        STDOUT_STREAM_FORWARD_TO_CONSOLE,
-        STDOUT_STREAM_RUNNING,
-} StdoutStreamState;
-
 /* The different types of log record terminators: a real \n was read, a NUL character was read, the maximum line length
  * was reached, or the end of the stream was reached */
 
@@ -72,44 +61,13 @@ typedef enum LineBreak {
         _LINE_BREAK_INVALID = -EINVAL,
 } LineBreak;
 
-struct StdoutStream {
-        Server *server;
-        StdoutStreamState state;
-
-        int fd;
-
-        struct ucred ucred;
-        char *label;
-        char *identifier;
-        char *unit_id;
-        int priority;
-        bool level_prefix:1;
-        bool forward_to_syslog:1;
-        bool forward_to_kmsg:1;
-        bool forward_to_console:1;
-
-        bool fdstore:1;
-        bool in_notify_queue:1;
-
-        char *buffer;
-        size_t length;
-
-        sd_event_source *event_source;
-
-        char *state_file;
-
-        ClientContext *context;
-
-        LIST_FIELDS(StdoutStream, stdout_stream);
-        LIST_FIELDS(StdoutStream, stdout_stream_notify_queue);
-
-        char id_field[STRLEN("_STREAM_ID=") + SD_ID128_STRING_MAX];
-};
-
 StdoutStream* stdout_stream_free(StdoutStream *s) {
         if (!s)
                 return NULL;
 
+        while (s->stream_sync_reqs)
+                stream_sync_req_free(s->stream_sync_reqs);
+
         if (s->server) {
                 if (s->context)
                         client_context_release(s->server, s->context);
@@ -120,8 +78,6 @@ StdoutStream* stdout_stream_free(StdoutStream *s) {
 
                 if (s->in_notify_queue)
                         LIST_REMOVE(stdout_stream_notify_queue, s->server->stdout_streams_notify_queue, s);
-
-                (void) server_start_or_stop_idle_timer(s->server); /* Maybe we are idle now? */
         }
 
         sd_event_source_disable_unref(s->event_source);
@@ -144,7 +100,16 @@ void stdout_stream_terminate(StdoutStream *s) {
         if (s->state_file)
                 (void) unlink(s->state_file);
 
-        stdout_stream_free(s);
+        StreamSyncReq *ssr;
+        while ((ssr = s->stream_sync_reqs)) {
+                SyncReq *req = ssr->req;
+                stream_sync_req_free(TAKE_PTR(ssr));
+                sync_req_revalidate(TAKE_PTR(req));
+        }
+
+        Server *server = s->server;
+        stdout_stream_free(TAKE_PTR(s));
+        (void) server_start_or_stop_idle_timer(server); /* Maybe we are idle now? */
 }
 
 static int stdout_stream_save(StdoutStream *s) {
@@ -646,6 +611,10 @@ static int stdout_stream_process(sd_event_source *es, int fd, uint32_t revents,
         s->length = l - consumed;
         memmove(s->buffer, p + consumed, s->length);
 
+        LIST_FOREACH(by_stdout_stream, ssr, s->stream_sync_reqs)
+                /* NB: this might invalidate the stdout stream! */
+                stream_sync_req_advance_revalidate(ssr, consumed);
+
         return 1;
 
 terminate:
@@ -746,14 +715,23 @@ static int stdout_stream_new(sd_event_source *es, int listen_fd, uint32_t revent
                 fd = safe_close(fd);
 
                 server_driver_message(s, u.pid, LOG_MESSAGE("Too many stdout streams, refusing connection."));
+
+                server_notify_stream(s, /* stream= */ NULL);
                 return 0;
         }
 
-        r = stdout_stream_install(s, fd, NULL);
-        if (r < 0)
+        StdoutStream *stream;
+        r = stdout_stream_install(s, fd, &stream);
+        if (r < 0) {
+                server_notify_stream(s, /* stream= */ NULL);
                 return r;
+        }
 
         TAKE_FD(fd);
+
+        /* Tell the synchronization logic that we dropped one item from the incoming connection queue */
+        server_notify_stream(s, stream);
+
         return 0;
 }
 
@@ -839,10 +817,9 @@ static int stdout_stream_restore(Server *s, const char *fname, int fd) {
         assert(fname);
         assert(fd >= 0);
 
-        if (s->n_stdout_streams >= STDOUT_STREAMS_MAX) {
-                log_warning("Too many stdout streams, refusing restoring of stream.");
-                return -ENOBUFS;
-        }
+        if (s->n_stdout_streams >= STDOUT_STREAMS_MAX)
+                return log_warning_errno(SYNTHETIC_ERRNO(ENOBUFS),
+                                         "Too many stdout streams, refusing restoring of stream.");
 
         r = stdout_stream_install(s, fd, &stream);
         if (r < 0)

src/journal/journald-stream.h

@@ -5,6 +5,54 @@ typedef struct StdoutStream StdoutStream;
 
 #include "fdset.h"
 #include "journald-server.h"
+#include "journald-sync.h"
+
+typedef enum StdoutStreamState {
+        STDOUT_STREAM_IDENTIFIER,
+        STDOUT_STREAM_UNIT_ID,
+        STDOUT_STREAM_PRIORITY,
+        STDOUT_STREAM_LEVEL_PREFIX,
+        STDOUT_STREAM_FORWARD_TO_SYSLOG,
+        STDOUT_STREAM_FORWARD_TO_KMSG,
+        STDOUT_STREAM_FORWARD_TO_CONSOLE,
+        STDOUT_STREAM_RUNNING,
+} StdoutStreamState;
+
+struct StdoutStream {
+        Server *server;
+        StdoutStreamState state;
+
+        int fd;
+
+        struct ucred ucred;
+        char *label;
+        char *identifier;
+        char *unit_id;
+        int priority;
+        bool level_prefix:1;
+        bool forward_to_syslog:1;
+        bool forward_to_kmsg:1;
+        bool forward_to_console:1;
+
+        bool fdstore:1;
+        bool in_notify_queue:1;
+
+        char *buffer;
+        size_t length;
+
+        sd_event_source *event_source;
+
+        char *state_file;
+
+        ClientContext *context;
+
+        LIST_FIELDS(StdoutStream, stdout_stream);
+        LIST_FIELDS(StdoutStream, stdout_stream_notify_queue);
+
+        char id_field[STRLEN("_STREAM_ID=") + SD_ID128_STRING_MAX];
+
+        LIST_HEAD(StreamSyncReq, stream_sync_reqs);
+};
 
 int server_open_stdout_socket(Server *s, const char *stdout_socket);
 int server_restore_streams(Server *s, FDSet *fds);

src/journal/journald-sync.c

@@ -0,0 +1,327 @@
+/* SPDX-License-Identifier: LGPL-2.1-or-later */
+
+#include <linux/sockios.h>
+#include <sys/ioctl.h>
+
+#include "sd-varlink.h"
+
+#include "journald-server.h"
+#include "journald-stream.h"
+#include "journald-sync.h"
+#include "journald-varlink.h"
+#include "socket-netlink.h"
+#include "time-util.h"
+
+StreamSyncReq *stream_sync_req_free(StreamSyncReq *ssr) {
+        if (!ssr)
+                return NULL;
+
+        if (ssr->req)
+                LIST_REMOVE(by_sync_req, ssr->req->stream_sync_reqs, ssr);
+        if (ssr->stream)
+                LIST_REMOVE(by_stdout_stream, ssr->stream->stream_sync_reqs, ssr);
+
+        return mfree(ssr);
+}
+
+void stream_sync_req_advance_revalidate(StreamSyncReq *ssr, size_t p) {
+        assert(ssr);
+
+        /* Subtract the specified number of bytes from the byte counter. And when we hit zero we consider
+         * this stream processed for the synchronization request */
+
+        /* NB: This might invalidate the 'ssr' object! */
+
+        if (p < ssr->pending_siocinq) {
+                ssr->pending_siocinq -= p;
+                return;
+        }
+
+        SyncReq *req = ASSERT_PTR(ssr->req);
+        stream_sync_req_free(TAKE_PTR(ssr));
+
+        /* Maybe we are done now? */
+        sync_req_revalidate(TAKE_PTR(req));
+}
+
+static bool sync_req_is_complete(SyncReq *req) {
+        int r;
+
+        assert(req);
+        assert(req->server);
+
+        if (req->prioq_idx != PRIOQ_IDX_NULL) {
+                /* If this sync request is still in the priority queue it means we still need to check if
+                 * incoming message timestamps are now newer than then sync request timestamp. */
+
+                if (req->server->native_event_source) {
+                        uint32_t revents = 0;
+
+                        r = sd_event_source_get_io_revents(req->server->native_event_source, &revents);
+                        if (r < 0 && r != -ENODATA)
+                                log_debug_errno(r, "Failed to determine pending IO events of native socket, ignoring: %m");
+
+                        if (FLAGS_SET(revents, EPOLLIN) &&
+                            req->server->native_timestamp < req->timestamp)
+                                return false;
+                }
+
+                if (req->server->syslog_event_source) {
+                        uint32_t revents = 0;
+
+                        r = sd_event_source_get_io_revents(req->server->syslog_event_source, &revents);
+                        if (r < 0 && r != -ENODATA)
+                                log_debug_errno(r, "Failed to determine pending IO events of syslog socket, ignoring: %m");
+
+                        if (FLAGS_SET(revents, EPOLLIN) &&
+                            req->server->syslog_timestamp < req->timestamp)
+                                return false;
+                }
+
+                /* This sync request is fulfilled for the native + syslog datagram streams? Then, let's
+                 * remove this sync request from the priority queue, so that we dont need to consider it
+                 * anymore. */
+                assert(prioq_remove(req->server->sync_req_prioq, req, &req->prioq_idx) > 0);
+        }
+
+        /* If there are still streams with pending counters, we still need to look into things */
+        if (req->stream_sync_reqs)
+                return false;
+
+        /* If there are still pending connections from before the sync started, we still need to look into things */
+        if (req->pending_rqlen > 0)
+                return false;
+
+        return true;
+}
+
+static int on_idle(sd_event_source *s, void *userdata) {
+        SyncReq *req = ASSERT_PTR(userdata);
+
+        req->idle_event_source = sd_event_source_disable_unref(req->idle_event_source);
+
+        /* When this idle event triggers, then we definitely are done with the synchronization request. This
+         * is a safety net of a kind, to ensure we'll definitely put an end to any synchronization request,
+         * even if we are confused by CLOCK_REALTIME jumps or similar. */
+        sync_req_varlink_reply(TAKE_PTR(req));
+        return 0;
+}
+
+SyncReq* sync_req_free(SyncReq *req) {
+        if (!req)
+                return NULL;
+
+        if (req->server) {
+                if (req->prioq_idx != PRIOQ_IDX_NULL)
+                        assert_se(prioq_remove(req->server->sync_req_prioq, req, &req->prioq_idx) > 0);
+
+                if (req->pending_rqlen > 0)
+                        LIST_REMOVE(pending_rqlen, req->server->sync_req_pending_rqlen, req);
+        }
+
+        req->idle_event_source = sd_event_source_disable_unref(req->idle_event_source);
+
+        sd_varlink_unref(req->link);
+
+        while (req->stream_sync_reqs)
+                stream_sync_req_free(req->stream_sync_reqs);
+
+        return mfree(req);
+}
+
+static int sync_req_compare(const void *a, const void *b) {
+        const SyncReq *x = a, *y = b;
+
+        return CMP(x->timestamp, y->timestamp);
+}
+
+static int sync_req_add_stream(SyncReq *req, StdoutStream *ss) {
+        assert(req);
+        assert(ss);
+
+        int v = 0;
+        if (ioctl(ss->fd, SIOCINQ, &v) < 0)
+                log_debug_errno(errno, "Failed to issue SIOCINQ on stream socket, ignoring: %m");
+        if (v <= 0)
+                return 0; /* Pending messages are zero anyway? then there's nothing to track */
+
+        _cleanup_(stream_sync_req_freep) StreamSyncReq *ssr = new(StreamSyncReq, 1);
+        if (!ssr)
+                return -ENOMEM;
+
+        *ssr = (StreamSyncReq) {
+                .stream = ss,
+                .pending_siocinq = v,
+                .req = req,
+        };
+
+        LIST_PREPEND(by_sync_req, req->stream_sync_reqs, ssr);
+        LIST_PREPEND(by_stdout_stream, ss->stream_sync_reqs, ssr);
+
+        TAKE_PTR(ssr);
+        return 1;
+}
+
+int sync_req_new(Server *s, sd_varlink *link, SyncReq **ret) {
+        int r;
+
+        assert(s);
+        assert(link);
+        assert(ret);
+
+        _cleanup_(sync_req_freep) SyncReq *req = new(SyncReq, 1);
+        if (!req)
+                return -ENOMEM;
+
+        *req = (SyncReq) {
+                .server = s,
+                .link = sd_varlink_ref(link),
+                .prioq_idx = PRIOQ_IDX_NULL,
+        };
+
+        /* We use three distinct mechanism to determine when the synchronization request is complete:
+         *
+         * 1. For the syslog/native AF_UNIX/SOCK_DGRAM sockets we look at the datagram timestamps: once the
+         *    most recently seen datagram on the socket is newer than the timestamp when we initiated the
+         *    requested we know that all previously enqueued messages have been processed by us.
+         *
+         * 2. For the stream AF_UNIX/SOCK_STREAM sockets we have no timestamps. For them we take the SIOCINQ
+         *    counter at the moment the synchronization request was enqueued. And once we processed the
+         *    indicated number of input bytes we know that anything further was enqueued later than the
+         *    original synchronization request timestamp we started from.
+         *
+         * 3. Finally, as safety net we install an idle handler with a very low priority (lower than the
+         *    syslog/native/stream IO handlers). If this handler is called we know that there's no pending
+         *    IO, hence everything so far queued is definitely processed.
+         *
+         * Note the asymmetry: for AF_UNIX/SOCK_DGRAM we go by timestamp, for AF_UNIX/SOCK_STREAM we count
+         * bytes. That's because for SOCK_STREAM we have no timestamps, and for SOCK_DGRAM we have no API to
+         * query all pending bytes (as SIOCINQ on SOCK_DGRAM reports size of next datagram, not size of all
+         * pending datagrams). Ideally, we'd actually use neither of this, and the kernel would provide us
+         * CLOCK_MONOTONIC timestamps...
+         *
+         * Note that CLOCK_REALTIME is not necessarily monotonic (that's the whole point of it after all). If
+         * the clock jumps then we know the algorithm will eventually terminate, because of the idle handler
+         * that is our safety net. */
+
+        req->timestamp = now(CLOCK_REALTIME);
+
+        r = prioq_ensure_put(&s->sync_req_prioq, sync_req_compare, req, &req->prioq_idx);
+        if (r < 0)
+                return r;
+
+        r = sd_event_add_defer(s->event, &req->idle_event_source, on_idle, req);
+        if (r < 0)
+                return r;
+
+        r = sd_event_source_set_priority(req->idle_event_source, SD_EVENT_PRIORITY_NORMAL+15);
+        if (r < 0)
+                return r;
+
+        (void) sd_event_source_set_description(req->idle_event_source, "deferred-sync");
+
+        /* Now determine the pending byte counter for each stdout stream. If non-zero allocate a
+         * StreamSyncReq for the stream to keep track of it */
+        LIST_FOREACH(stdout_stream, ss, s->stdout_streams) {
+                r = sync_req_add_stream(req, ss);
+                if (r < 0)
+                        return r;
+        }
+
+        /* Also track how many pending, incoming stream sockets there are currently, so that we process them
+         * too */
+        r = af_unix_get_qlen(s->stdout_fd, &req->pending_rqlen);
+        if (r < 0)
+                log_warning_errno(r, "Failed to determine current incoming queue length, ignoring: %m");
+        if (req->pending_rqlen > 0)
+                LIST_PREPEND(pending_rqlen, s->sync_req_pending_rqlen, req);
+
+        *ret = TAKE_PTR(req);
+        return 0;
+}
+
+static void sync_req_advance_rqlen_revalidate(SyncReq *req, uint32_t current_rqlen, StdoutStream *ss) {
+        int r;
+
+        assert(req);
+
+        /* Invoked whenever a new connection was accept()ed, i.e. dropped off the queue of pending incoming
+         * connections. We decrease the qlen counter by one here, except if the new overall counter is
+         * already below our target. */
+
+        uint32_t n;
+        if (req->pending_rqlen <= 0)
+                n = 0;
+        else if (req->pending_rqlen > current_rqlen)
+                n = current_rqlen;
+        else
+                n = req->pending_rqlen - 1;
+
+        if (req->pending_rqlen > 0) {
+                /* if this synchronization request is supposed to process a non-zero number of connections we
+                 * need to also track what's inside those stream connections */
+                if (ss) {
+                        r = sync_req_add_stream(req, ss);
+                        if (r < 0)
+                                log_warning_errno(r, "Failed to track stream queue size, ignoring: %m");
+                }
+
+                /* If there are no more connections to wait for, remove us from the list of synchronization
+                 * requests with non-zero pending connection counters */
+                if (n == 0)
+                        LIST_REMOVE(pending_rqlen, req->server->sync_req_pending_rqlen, req);
+        }
+
+        req->pending_rqlen = n;
+
+        sync_req_revalidate(req);
+}
+
+void server_notify_stream(Server *s, StdoutStream *ss) {
+        int r;
+
+        assert(s);
+
+        /* Invoked whenever a new connection was accept()ed, i.e. dropped off the queue of pending incoming
+         * connections. */
+
+        if (!s->sync_req_pending_rqlen)
+                return;
+
+        uint32_t current_qlen;
+
+        r = af_unix_get_qlen(s->stdout_fd, &current_qlen);
+        if (r < 0) {
+                log_warning_errno(r, "Failed to determine current AF_UNIX stream socket pending connections, ignoring: %m");
+                current_qlen = UINT32_MAX;
+        }
+
+        LIST_FOREACH(pending_rqlen, sr, s->sync_req_pending_rqlen)
+                /* NB: this might invalidate the SyncReq object! */
+                sync_req_advance_rqlen_revalidate(sr, current_qlen, ss);
+}
+
+bool sync_req_revalidate(SyncReq *req) {
+        assert(req);
+
+        /* Check if the synchronization request is complete now. If so, answer the Varlink client. NB: this
+         * might invalidate the SyncReq object */
+
+        if (!sync_req_is_complete(req))
+                return false;
+
+        sync_req_varlink_reply(TAKE_PTR(req));
+        return true;
+}
+
+void sync_req_revalidate_by_timestamp(Server *s) {
+        assert(s);
+
+        /* Go through the pending sync requests by timestamp, and complete those for which a sync is now
+         * complete. */
+
+        SyncReq *req;
+        while ((req = prioq_peek(s->sync_req_prioq)))
+                if (!sync_req_revalidate(req))
+                        break;
+}

src/journal/journald-sync.h

@@ -0,0 +1,51 @@
+/* SPDX-License-Identifier: LGPL-2.1-or-later */
+#pragma once
+
+typedef struct SyncReq SyncReq;
+typedef struct StreamSyncReq StreamSyncReq;
+
+#include "journald-server.h"
+#include "macro.h"
+
+/* Encapsulates the synchronization request data we need to keep per STDOUT stream. Primarily a byte counter
+ * to count down. */
+struct StreamSyncReq {
+        SyncReq *req;
+        StdoutStream *stream;
+
+        uint64_t pending_siocinq; /* The SIOCINQ counter when the sync was initiated */
+
+        LIST_FIELDS(StreamSyncReq, by_sync_req);
+        LIST_FIELDS(StreamSyncReq, by_stdout_stream);
+};
+
+/* Encapsulates a synchronization request */
+struct SyncReq {
+        Server *server;
+        sd_varlink *link;
+
+        bool offline; /* if true, we'll offline the journal files after sync is complete */
+
+        usec_t timestamp; /* CLOCK_REALTIME timestamp when synchronization request was initiated */
+        sd_event_source *idle_event_source;
+
+        uint32_t pending_rqlen;   /* The rqlen counter on the stream AF_UNIX socket when the sync was initiated */
+        LIST_FIELDS(SyncReq, pending_rqlen);
+
+        LIST_HEAD(StreamSyncReq, stream_sync_reqs);
+
+        unsigned prioq_idx;
+};
+
+StreamSyncReq *stream_sync_req_free(StreamSyncReq *ssr);
+DEFINE_TRIVIAL_CLEANUP_FUNC(StreamSyncReq*, stream_sync_req_free);
+void stream_sync_req_advance_revalidate(StreamSyncReq *ssr, size_t p);
+
+int sync_req_new(Server *s, sd_varlink *link, SyncReq **ret);
+SyncReq* sync_req_free(SyncReq *req);
+DEFINE_TRIVIAL_CLEANUP_FUNC(SyncReq*, sync_req_free);
+
+bool sync_req_revalidate(SyncReq *req);
+void sync_req_revalidate_by_timestamp(Server *s);
+
+void server_notify_stream(Server *s, StdoutStream *ss);

src/journal/journald-varlink.c

@@ -1,75 +1,78 @@
 /* SPDX-License-Identifier: LGPL-2.1-or-later */
 
+#include "journald-sync.h"
 #include "journald-varlink.h"
 #include "varlink-io.systemd.Journal.h"
 #include "varlink-io.systemd.service.h"
 #include "varlink-util.h"
 
-static int synchronize_second_half(sd_event_source *event_source, void *userdata) {
-        sd_varlink *link = ASSERT_PTR(userdata);
-        Server *s;
+void sync_req_varlink_reply(SyncReq *req) {
         int r;
 
-        assert_se(s = sd_varlink_get_userdata(link));
+        assert(req);
 
-        /* This is the "second half" of the Synchronize() varlink method. This function is called as deferred
-         * event source at a low priority to ensure the synchronization completes after all queued log
-         * messages are processed. */
-        server_full_sync(s, /* wait = */ true);
+        /* This is the "second half" of the Synchronize() varlink method. This function is called when we
+         * determine that no messages that were enqueued to us when the request was initiated is pending
+         * anymore. */
 
-        /* Let's get rid of the event source now, by marking it as non-floating again. It then has no ref
-         * anymore and is immediately destroyed after we return from this function, i.e. from this event
-         * source handler at the end. */
-        r = sd_event_source_set_floating(event_source, false);
+        if (req->offline)
+                server_full_sync(req->server, /* wait = */ true);
+
+        /* Disconnect the SyncReq from the Varlink connection object, and free it */
+        _cleanup_(sd_varlink_unrefp) sd_varlink *vl = TAKE_PTR(req->link);
+        sd_varlink_set_userdata(vl, req->server); /* reinstall server object */
+        req = sync_req_free(req);
+
+        r = sd_varlink_reply(vl, NULL);
         if (r < 0)
-                return log_error_errno(r, "Failed to mark event source as non-floating: %m");
-
-        return sd_varlink_reply(link, NULL);
-}
-
-static void synchronize_destroy(void *userdata) {
-        sd_varlink_unref(userdata);
+                log_debug_errno(r, "Failed to reply to Synchronize() client, ignoring: %m");
 }
 
 static int vl_method_synchronize(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata) {
-        _cleanup_(sd_event_source_unrefp) sd_event_source *event_source = NULL;
+        int offline = -1;
+
+        static const sd_json_dispatch_field dispatch_table[] = {
+                { "offline", SD_JSON_VARIANT_BOOLEAN, sd_json_dispatch_tristate, 0, 0},
+                {}
+        };
+
         Server *s = ASSERT_PTR(userdata);
         int r;
 
         assert(link);
 
-        r = sd_varlink_dispatch(link, parameters, /* dispatch_table = */ NULL, /* userdata = */ NULL);
+        r = sd_varlink_dispatch(link, parameters, dispatch_table, &offline);
         if (r != 0)
                 return r;
 
-        log_info("Received client request to sync journal.");
+        if (offline > 0) {
+                /* Do not allow unprivileged clients to offline the journal files, since that's potentially slow */
+                r = varlink_check_privileged_peer(link);
+                if (r < 0)
+                        return r;
+        } else if (offline < 0) {
+                uid_t uid = 0;
 
-        /* We don't do the main work now, but instead enqueue a deferred event loop job which will do
-         * it. That job is scheduled at low priority, so that we return from this method call only after all
-         * queued but not processed log messages are written to disk, so that this method call returning can
-         * be used as nice synchronization point. */
-        r = sd_event_add_defer(s->event, &event_source, synchronize_second_half, link);
+                r = sd_varlink_get_peer_uid(link, &uid);
+                if (r < 0)
+                        return r;
+
+                offline = uid == 0; /* for compat, if not specified default to offlining, except for non-root */
+        }
+
+        log_full(offline ? LOG_INFO : LOG_DEBUG,
+                 "Received client request to sync journal (%s offlining).", offline ? "with" : "without");
+
+        _cleanup_(sync_req_freep) SyncReq *sr = NULL;
+
+        r = sync_req_new(s, link, &sr);
         if (r < 0)
-                return log_error_errno(r, "Failed to allocate defer event source: %m");
+                return r;
 
-        r = sd_event_source_set_destroy_callback(event_source, synchronize_destroy);
-        if (r < 0)
-                return log_error_errno(r, "Failed to set event source destroy callback: %m");
-
-        sd_varlink_ref(link); /* The varlink object is now left to the destroy callback to unref */
-
-        r = sd_event_source_set_priority(event_source, SD_EVENT_PRIORITY_NORMAL+15);
-        if (r < 0)
-                return log_error_errno(r, "Failed to set defer event source priority: %m");
-
-        /* Give up ownership of this event source. It will now be destroyed along with event loop itself,
-         * unless it destroys itself earlier. */
-        r = sd_event_source_set_floating(event_source, true);
-        if (r < 0)
-                return log_error_errno(r, "Failed to mark event source as floating: %m");
-
-        (void) sd_event_source_set_description(event_source, "deferred-sync");
+        sr->offline = offline;
+        sd_varlink_set_userdata(link, sr);
 
+        sync_req_revalidate(TAKE_PTR(sr));
         return 0;
 }
 
@@ -83,6 +86,10 @@ static int vl_method_rotate(sd_varlink *link, sd_json_variant *parameters, sd_va
         if (r != 0)
                 return r;
 
+        r = varlink_check_privileged_peer(link);
+        if (r < 0)
+                return r;
+
         log_info("Received client request to rotate journal, rotating.");
         server_full_rotate(s);
 
@@ -99,6 +106,10 @@ static int vl_method_flush_to_var(sd_varlink *link, sd_json_variant *parameters,
         if (r != 0)
                 return r;
 
+        r = varlink_check_privileged_peer(link);
+        if (r < 0)
+                return r;
+
         if (s->namespace)
                 return sd_varlink_error(link, "io.systemd.Journal.NotSupportedByNamespaces", NULL);
 
@@ -118,6 +129,10 @@ static int vl_method_relinquish_var(sd_varlink *link, sd_json_variant *parameter
         if (r != 0)
                 return r;
 
+        r = varlink_check_privileged_peer(link);
+        if (r < 0)
+                return r;
+
         if (s->namespace)
                 return sd_varlink_error(link, "io.systemd.Journal.NotSupportedByNamespaces", NULL);
 
@@ -144,6 +159,15 @@ static void vl_disconnect(sd_varlink_server *server, sd_varlink *link, void *use
         assert(server);
         assert(link);
 
+        void *u = sd_varlink_get_userdata(link);
+        if (u != s) {
+                /* If this is a Varlink connection that does not have the Server object as userdata, then it has a SyncReq object instead. Let's finish it. */
+
+                SyncReq *req = u;
+                sd_varlink_set_userdata(link, s); /* reinstall server object */
+                sync_req_free(req);
+        }
+
         (void) server_start_or_stop_idle_timer(s); /* maybe we are idle now */
 }
 
@@ -154,7 +178,7 @@ int server_open_varlink(Server *s, const char *socket, int fd) {
 
         r = varlink_server_new(
                         &s->varlink_server,
-                        SD_VARLINK_SERVER_ROOT_ONLY|SD_VARLINK_SERVER_INHERIT_USERDATA,
+                        SD_VARLINK_SERVER_ACCOUNT_UID|SD_VARLINK_SERVER_INHERIT_USERDATA,
                         s);
         if (r < 0)
                 return log_error_errno(r, "Failed to allocate varlink server object: %m");
@@ -187,7 +211,7 @@ int server_open_varlink(Server *s, const char *socket, int fd) {
                 return r;
 
         if (fd < 0)
-                r = sd_varlink_server_listen_address(s->varlink_server, socket, 0600);
+                r = sd_varlink_server_listen_address(s->varlink_server, socket, 0666);
         else
                 r = sd_varlink_server_listen_fd(s->varlink_server, fd);
         if (r < 0)

src/journal/journald-varlink.h

@@ -2,5 +2,8 @@
 #pragma once
 
 #include "journald-server.h"
+#include "journald-sync.h"
 
 int server_open_varlink(Server *s, const char *socket, int fd);
+
+void sync_req_varlink_reply(SyncReq *req);

src/journal/meson.build

@@ -11,6 +11,7 @@ sources = files(
         'journald-server.c',
         'journald-socket.c',
         'journald-stream.c',
+        'journald-sync.c',
         'journald-syslog.c',
         'journald-varlink.c',
         'journald-wall.c',

src/libsystemd/meson.build

@@ -131,10 +131,12 @@ sd_netlink_sources = files(
         'sd-netlink/netlink-message-rtnl.c',
         'sd-netlink/netlink-message.c',
         'sd-netlink/netlink-slot.c',
+        'sd-netlink/netlink-sock-diag.c',
         'sd-netlink/netlink-socket.c',
         'sd-netlink/netlink-types-genl.c',
         'sd-netlink/netlink-types-nfnl.c',
         'sd-netlink/netlink-types-rtnl.c',
+        'sd-netlink/netlink-types-sdnl.c',
         'sd-netlink/netlink-types.c',
         'sd-netlink/netlink-util.c',
         'sd-netlink/sd-netlink.c',

src/libsystemd/sd-journal/journal-file.c

@@ -98,7 +98,7 @@ DEFINE_HASH_OPS_WITH_VALUE_DESTRUCTOR(
                 JournalFile, journal_file_close);
 
 static int mmap_prot_from_open_flags(int flags) {
-        switch (flags & O_ACCMODE) {
+        switch (flags & O_ACCMODE_STRICT) {
         case O_RDONLY:
                 return PROT_READ;
         case O_WRONLY:
@@ -4075,10 +4075,10 @@ int journal_file_open(
         assert(mmap_cache);
         assert(ret);
 
-        if (!IN_SET((open_flags & O_ACCMODE), O_RDONLY, O_RDWR))
+        if (!IN_SET((open_flags & O_ACCMODE_STRICT), O_RDONLY, O_RDWR))
                 return -EINVAL;
 
-        if ((open_flags & O_ACCMODE) == O_RDONLY && FLAGS_SET(open_flags, O_CREAT))
+        if ((open_flags & O_ACCMODE_STRICT) == O_RDONLY && FLAGS_SET(open_flags, O_CREAT))
                 return -EINVAL;
 
         if (fname && (open_flags & O_CREAT) && !endswith(fname, ".journal"))

src/libsystemd/sd-journal/journal-file.h

@@ -1,7 +1,6 @@
 /* SPDX-License-Identifier: LGPL-2.1-or-later */
 #pragma once
 
-#include <fcntl.h>
 #include <inttypes.h>
 #include <sys/uio.h>
 
@@ -15,6 +14,7 @@
 #include "compress.h"
 #include "hashmap.h"
 #include "journal-def.h"
+#include "missing_fcntl.h"
 #include "mmap-cache.h"
 #include "sparse-endian.h"
 #include "time-util.h"
@@ -391,5 +391,5 @@ static inline uint32_t COMPRESSION_TO_HEADER_INCOMPATIBLE_FLAG(Compression c) {
 
 static inline bool journal_file_writable(JournalFile *f) {
         assert(f);
-        return (f->open_flags & O_ACCMODE) != O_RDONLY;
+        return (f->open_flags & O_ACCMODE_STRICT) != O_RDONLY;
 }

src/libsystemd/sd-netlink/netlink-genl.c

@@ -233,7 +233,7 @@ static int genl_message_new(
         if (!policy_set)
                 return -EOPNOTSUPP;
 
-        r = message_new_full(nl, family->id, policy_set,
+        r = message_new_full(nl, family->id, NLM_F_REQUEST | NLM_F_ACK, policy_set,
                              sizeof(struct genlmsghdr) + family->additional_header_size, &m);
         if (r < 0)
                 return r;

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

@@ -131,10 +131,11 @@ int message_new_empty(sd_netlink *nl, sd_netlink_message **ret);
 int message_new_full(
                 sd_netlink *nl,
                 uint16_t nlmsg_type,
+                uint16_t nlmsg_flags,
                 const NLAPolicySet *policy_set,
                 size_t header_size,
                 sd_netlink_message **ret);
-int message_new(sd_netlink *nl, sd_netlink_message **ret, uint16_t type);
+int message_new(sd_netlink *nl, sd_netlink_message **ret, uint16_t type, uint16_t flags);
 int message_new_synthetic_error(sd_netlink *nl, int error, uint32_t serial, sd_netlink_message **ret);
 
 static inline uint32_t message_get_serial(sd_netlink_message *m) {

src/libsystemd/sd-netlink/netlink-message-nfnl.c

@@ -32,7 +32,7 @@ int sd_nfnl_message_new(sd_netlink *nfnl, sd_netlink_message **ret, int nfproto,
         assert_return(nfproto_is_valid(nfproto), -EINVAL);
         assert_return(NFNL_MSG_TYPE(msg_type) == msg_type, -EINVAL);
 
-        r = message_new(nfnl, &m, subsys << 8 | msg_type);
+        r = message_new(nfnl, &m, subsys << 8 | msg_type, NLM_F_REQUEST | NLM_F_ACK);
         if (r < 0)
                 return r;
 

src/libsystemd/sd-netlink/netlink-message-rtnl.c

@@ -243,13 +243,13 @@ int sd_rtnl_message_new_route(
                       IN_SET(family, AF_INET, AF_INET6), -EINVAL);
         assert_return(ret, -EINVAL);
 
-        r = message_new(rtnl, ret, nlmsg_type);
+        r = message_new(rtnl,
+                        ret,
+                        nlmsg_type,
+                        NLM_F_REQUEST|NLM_F_ACK|(nlmsg_type == RTM_NEWROUTE ? NLM_F_CREATE | NLM_F_APPEND : 0));
         if (r < 0)
                 return r;
 
-        if (nlmsg_type == RTM_NEWROUTE)
-                (*ret)->hdr->nlmsg_flags |= NLM_F_CREATE | NLM_F_APPEND;
-
         rtm = NLMSG_DATA((*ret)->hdr);
 
         rtm->rtm_family = family;
@@ -280,13 +280,13 @@ int sd_rtnl_message_new_nexthop(sd_netlink *rtnl, sd_netlink_message **ret,
         }
         assert_return(ret, -EINVAL);
 
-        r = message_new(rtnl, ret, nlmsg_type);
+        r = message_new(rtnl,
+                        ret,
+                        nlmsg_type,
+                        NLM_F_REQUEST | NLM_F_ACK | (nlmsg_type == RTM_NEWNEXTHOP ? NLM_F_CREATE | NLM_F_REPLACE : 0));
         if (r < 0)
                 return r;
 
-        if (nlmsg_type == RTM_NEWNEXTHOP)
-                (*ret)->hdr->nlmsg_flags |= NLM_F_CREATE | NLM_F_REPLACE;
-
         nhm = NLMSG_DATA((*ret)->hdr);
 
         nhm->nh_family = family;
@@ -310,17 +310,18 @@ int sd_rtnl_message_new_neigh(
         assert_return(IN_SET(family, AF_UNSPEC, AF_INET, AF_INET6, AF_BRIDGE), -EINVAL);
         assert_return(ret, -EINVAL);
 
-        r = message_new(rtnl, ret, nlmsg_type);
-        if (r < 0)
-                return r;
-
+        uint16_t flags = NLM_F_REQUEST | NLM_F_ACK;
         if (nlmsg_type == RTM_NEWNEIGH) {
                 if (family == AF_BRIDGE)
-                        (*ret)->hdr->nlmsg_flags |= NLM_F_CREATE | NLM_F_APPEND;
+                        flags |= NLM_F_CREATE | NLM_F_APPEND;
                 else
-                        (*ret)->hdr->nlmsg_flags |= NLM_F_CREATE | NLM_F_REPLACE;
+                        flags |= NLM_F_CREATE | NLM_F_REPLACE;
         }
 
+        r = message_new(rtnl, ret, nlmsg_type, flags);
+        if (r < 0)
+                return r;
+
         ndm = NLMSG_DATA((*ret)->hdr);
 
         ndm->ndm_family = family;
@@ -336,15 +337,16 @@ int sd_rtnl_message_new_link(sd_netlink *rtnl, sd_netlink_message **ret, uint16_
         assert_return(rtnl_message_type_is_link(nlmsg_type), -EINVAL);
         assert_return(ret, -EINVAL);
 
-        r = message_new(rtnl, ret, nlmsg_type);
+        uint16_t flags = NLM_F_REQUEST | NLM_F_ACK;
+        if (nlmsg_type == RTM_NEWLINK && ifindex == 0)
+                flags |= NLM_F_CREATE | NLM_F_EXCL;
+        else if (nlmsg_type == RTM_NEWLINKPROP)
+                flags |= NLM_F_CREATE | NLM_F_EXCL | NLM_F_APPEND;
+
+        r = message_new(rtnl, ret, nlmsg_type, flags);
         if (r < 0)
                 return r;
 
-        if (nlmsg_type == RTM_NEWLINK && ifindex == 0)
-                (*ret)->hdr->nlmsg_flags |= NLM_F_CREATE | NLM_F_EXCL;
-        else if (nlmsg_type == RTM_NEWLINKPROP)
-                (*ret)->hdr->nlmsg_flags |= NLM_F_CREATE | NLM_F_EXCL | NLM_F_APPEND;
-
         ifi = NLMSG_DATA((*ret)->hdr);
 
         ifi->ifi_family = AF_UNSPEC;
@@ -370,7 +372,7 @@ int sd_rtnl_message_new_addr(
                       IN_SET(family, AF_INET, AF_INET6), -EINVAL);
         assert_return(ret, -EINVAL);
 
-        r = message_new(rtnl, ret, nlmsg_type);
+        r = message_new(rtnl, ret, nlmsg_type, NLM_F_REQUEST | NLM_F_ACK);
         if (r < 0)
                 return r;
 
@@ -438,13 +440,13 @@ int sd_rtnl_message_new_addrlabel(
         assert_return(rtnl_message_type_is_addrlabel(nlmsg_type), -EINVAL);
         assert_return(ret, -EINVAL);
 
-        r = message_new(rtnl, ret, nlmsg_type);
+        r = message_new(rtnl,
+                        ret,
+                        nlmsg_type,
+                        NLM_F_REQUEST | NLM_F_ACK | (nlmsg_type == RTM_NEWADDRLABEL ? NLM_F_CREATE | NLM_F_REPLACE : 0));
         if (r < 0)
                 return r;
 
-        if (nlmsg_type == RTM_NEWADDRLABEL)
-                (*ret)->hdr->nlmsg_flags |= NLM_F_CREATE | NLM_F_REPLACE;
-
         addrlabel = NLMSG_DATA((*ret)->hdr);
 
         addrlabel->ifal_family = family;
@@ -465,13 +467,13 @@ int sd_rtnl_message_new_routing_policy_rule(
         assert_return(rtnl_message_type_is_routing_policy_rule(nlmsg_type), -EINVAL);
         assert_return(ret, -EINVAL);
 
-        r = message_new(rtnl, ret, nlmsg_type);
+        r = message_new(rtnl,
+                        ret,
+                        nlmsg_type,
+                        NLM_F_REQUEST | NLM_F_ACK | (nlmsg_type == RTM_NEWRULE ? NLM_F_CREATE | NLM_F_EXCL : 0));
         if (r < 0)
                 return r;
 
-        if (nlmsg_type == RTM_NEWRULE)
-                (*ret)->hdr->nlmsg_flags |= NLM_F_CREATE | NLM_F_EXCL;
-
         frh = NLMSG_DATA((*ret)->hdr);
         frh->family = family;
 
@@ -492,13 +494,13 @@ int sd_rtnl_message_new_traffic_control(
         assert_return(rtnl_message_type_is_traffic_control(nlmsg_type), -EINVAL);
         assert_return(ret, -EINVAL);
 
-        r = message_new(rtnl, ret, nlmsg_type);
+        r = message_new(rtnl,
+                        ret,
+                        nlmsg_type,
+                        NLM_F_REQUEST | NLM_F_ACK | (IN_SET(nlmsg_type, RTM_NEWQDISC, RTM_NEWTCLASS) ? NLM_F_CREATE | NLM_F_REPLACE : 0));
         if (r < 0)
                 return r;
 
-        if (IN_SET(nlmsg_type, RTM_NEWQDISC, RTM_NEWTCLASS))
-                (*ret)->hdr->nlmsg_flags |= NLM_F_CREATE | NLM_F_REPLACE;
-
         tcm = NLMSG_DATA((*ret)->hdr);
         tcm->tcm_ifindex = ifindex;
         tcm->tcm_handle = handle;
@@ -519,13 +521,13 @@ int sd_rtnl_message_new_mdb(
         assert_return(rtnl_message_type_is_mdb(nlmsg_type), -EINVAL);
         assert_return(ret, -EINVAL);
 
-        r = message_new(rtnl, ret, nlmsg_type);
+        r = message_new(rtnl,
+                        ret,
+                        nlmsg_type,
+                        NLM_F_REQUEST | NLM_F_ACK | (nlmsg_type == RTM_NEWMDB ? NLM_F_CREATE | NLM_F_REPLACE : 0));
         if (r < 0)
                 return r;
 
-        if (nlmsg_type == RTM_NEWMDB)
-                (*ret)->hdr->nlmsg_flags |= NLM_F_CREATE | NLM_F_REPLACE;
-
         bpm = NLMSG_DATA((*ret)->hdr);
         bpm->family = AF_BRIDGE;
         bpm->ifindex = ifindex;
@@ -544,7 +546,7 @@ int sd_rtnl_message_new_nsid(
         assert_return(rtnl_message_type_is_nsid(nlmsg_type), -EINVAL);
         assert_return(ret, -EINVAL);
 
-        r = message_new(rtnl, ret, nlmsg_type);
+        r = message_new(rtnl, ret, nlmsg_type, NLM_F_REQUEST | NLM_F_ACK);
         if (r < 0)
                 return r;
 

src/libsystemd/sd-netlink/netlink-message.c

@@ -43,6 +43,7 @@ int message_new_empty(sd_netlink *nl, sd_netlink_message **ret) {
 int message_new_full(
                 sd_netlink *nl,
                 uint16_t nlmsg_type,
+                uint16_t nlmsg_flags,
                 const NLAPolicySet *policy_set,
                 size_t header_size,
                 sd_netlink_message **ret) {
@@ -68,7 +69,7 @@ int message_new_full(
         if (!m->hdr)
                 return -ENOMEM;
 
-        m->hdr->nlmsg_flags = NLM_F_REQUEST | NLM_F_ACK;
+        m->hdr->nlmsg_flags = nlmsg_flags;
         m->hdr->nlmsg_len = size;
         m->hdr->nlmsg_type = nlmsg_type;
 
@@ -76,7 +77,7 @@ int message_new_full(
         return 0;
 }
 
-int message_new(sd_netlink *nl, sd_netlink_message **ret, uint16_t nlmsg_type) {
+int message_new(sd_netlink *nl, sd_netlink_message **ret, uint16_t nlmsg_type, uint16_t nlmsg_flags) {
         const NLAPolicySet *policy_set;
         size_t size;
         int r;
@@ -84,11 +85,11 @@ int message_new(sd_netlink *nl, sd_netlink_message **ret, uint16_t nlmsg_type) {
         assert_return(nl, -EINVAL);
         assert_return(ret, -EINVAL);
 
-        r = netlink_get_policy_set_and_header_size(nl, nlmsg_type, &policy_set, &size);
+        r = netlink_get_policy_set_and_header_size(nl, nlmsg_type, nlmsg_flags, &policy_set, &size);
         if (r < 0)
                 return r;
 
-        return message_new_full(nl, nlmsg_type, policy_set, size, ret);
+        return message_new_full(nl, nlmsg_type, nlmsg_flags, policy_set, size, ret);
 }
 
 int message_new_synthetic_error(sd_netlink *nl, int error, uint32_t serial, sd_netlink_message **ret) {
@@ -97,7 +98,7 @@ int message_new_synthetic_error(sd_netlink *nl, int error, uint32_t serial, sd_n
 
         assert(error <= 0);
 
-        r = message_new(nl, ret, NLMSG_ERROR);
+        r = message_new(nl, ret, NLMSG_ERROR, 0);
         if (r < 0)
                 return r;
 
@@ -1327,8 +1328,12 @@ int sd_netlink_message_rewind(sd_netlink_message *m, sd_netlink *nl) {
 
         assert(m->hdr);
 
-        r = netlink_get_policy_set_and_header_size(nl, m->hdr->nlmsg_type,
-                                                   &m->containers[0].policy_set, &size);
+        r = netlink_get_policy_set_and_header_size(
+                        nl,
+                        m->hdr->nlmsg_type,
+                        m->hdr->nlmsg_flags,
+                        &m->containers[0].policy_set,
+                        &size);
         if (r < 0)
                 return r;
 

src/libsystemd/sd-netlink/netlink-sock-diag.c

@@ -0,0 +1,43 @@
+/* SPDX-License-Identifier: LGPL-2.1-or-later */
+
+#include <linux/sock_diag.h>
+#include <linux/unix_diag.h>
+
+#include "netlink-internal.h"
+#include "netlink-sock-diag.h"
+#include "netlink-util.h"
+
+int sd_sock_diag_socket_open(sd_netlink **ret) {
+        return netlink_open_family(ret, NETLINK_SOCK_DIAG);
+}
+
+int sd_sock_diag_message_new_unix(
+                sd_netlink *sdnl,
+                sd_netlink_message **ret,
+                ino_t inode,
+                uint64_t cookie,
+                uint32_t show) {
+
+        _cleanup_(sd_netlink_message_unrefp) sd_netlink_message *m = NULL;
+        int r;
+
+        assert_return(sdnl, -EINVAL);
+        assert_return(ret, -EINVAL);
+
+        r = message_new(sdnl, &m, SOCK_DIAG_BY_FAMILY, NLM_F_REQUEST | NLM_F_ACK);
+        if (r < 0)
+                return r;
+
+        *(struct unix_diag_req*) NLMSG_DATA(m->hdr) = (struct unix_diag_req) {
+                .sdiag_family = AF_UNIX,
+                .udiag_ino = inode,
+                .udiag_show = show,
+                .udiag_cookie = {
+                        cookie & UINT32_MAX,
+                        (cookie >> 32) & UINT32_MAX,
+                },
+        };
+
+        *ret = TAKE_PTR(m);
+        return 0;
+}

src/libsystemd/sd-netlink/netlink-sock-diag.h

@@ -0,0 +1,11 @@
+/* SPDX-License-Identifier: LGPL-2.1-or-later */
+#pragma once
+
+#include <sys/types.h>
+
+#include "sd-netlink.h"
+
+/* TODO: to be exported later */
+
+int sd_sock_diag_socket_open(sd_netlink **ret);
+int sd_sock_diag_message_new_unix(sd_netlink *sdnl, sd_netlink_message **ret, ino_t inode, uint64_t cookie, uint32_t show);

src/libsystemd/sd-netlink/netlink-socket.c

@@ -306,7 +306,7 @@ static int parse_message_one(sd_netlink *nl, uint32_t group, const struct nlmsgh
                 goto finalize;
 
         /* check that we support this message type */
-        r = netlink_get_policy_set_and_header_size(nl, hdr->nlmsg_type, NULL, &size);
+        r = netlink_get_policy_set_and_header_size(nl, hdr->nlmsg_type, hdr->nlmsg_flags, NULL, &size);
         if (r == -EOPNOTSUPP) {
                 log_debug("sd-netlink: ignored message with unknown type: %i", hdr->nlmsg_type);
                 goto finalize;

src/libsystemd/sd-netlink/netlink-types-sdnl.c

@@ -0,0 +1,37 @@
+/* SPDX-License-Identifier: LGPL-2.1-or-later */
+
+#include <linux/sock_diag.h>
+#include <linux/unix_diag.h>
+
+#include "missing_network.h"
+#include "netlink-types-internal.h"
+#include "netlink-types.h"
+
+static const NLAPolicy unix_diag_req_policies[] = {
+};
+DEFINE_POLICY_SET(unix_diag_req);
+
+static const NLAPolicy sdnl_req_policies[] = {
+        [SOCK_DIAG_BY_FAMILY] = BUILD_POLICY_NESTED_WITH_SIZE(unix_diag_req, sizeof(struct unix_diag_req)),
+};
+
+DEFINE_POLICY_SET(sdnl_req);
+
+static const NLAPolicy unix_diag_msg_policies[] = {
+        [UNIX_DIAG_RQLEN] = BUILD_POLICY_WITH_SIZE(BINARY, sizeof(struct unix_diag_rqlen)),
+};
+DEFINE_POLICY_SET(unix_diag_msg);
+
+static const NLAPolicy sdnl_msg_policies[] = {
+        [SOCK_DIAG_BY_FAMILY] = BUILD_POLICY_NESTED_WITH_SIZE(unix_diag_msg, sizeof(struct unix_diag_msg)),
+};
+
+DEFINE_POLICY_SET(sdnl_msg);
+
+const NLAPolicy *sdnl_get_policy(uint16_t nlmsg_type, uint16_t flags) {
+        /* for sock_diag we need to look at whether a message is a response or request to determine how to decode it. */
+        if (flags & NLM_F_REQUEST)
+                return policy_set_get_policy(&sdnl_req_policy_set, nlmsg_type);
+
+        return policy_set_get_policy(&sdnl_msg_policy_set, nlmsg_type);
+}

src/libsystemd/sd-netlink/netlink-types.c

@@ -51,6 +51,7 @@ const NLAPolicySetUnion *policy_get_policy_set_union(const NLAPolicy *policy) {
 int netlink_get_policy_set_and_header_size(
                 sd_netlink *nl,
                 uint16_t type,
+                uint16_t flags,
                 const NLAPolicySet **ret_policy_set,
                 size_t *ret_header_size) {
 
@@ -70,6 +71,9 @@ int netlink_get_policy_set_and_header_size(
                         break;
                 case NETLINK_GENERIC:
                         return genl_get_policy_set_and_header_size(nl, type, ret_policy_set, ret_header_size);
+                case NETLINK_SOCK_DIAG:
+                        policy = sdnl_get_policy(type, flags);
+                        break;
                 default:
                         return -EOPNOTSUPP;
                 }

src/libsystemd/sd-netlink/netlink-types.h

@@ -37,6 +37,7 @@ typedef struct NLAPolicySetUnion NLAPolicySetUnion;
 
 const NLAPolicy *rtnl_get_policy(uint16_t nlmsg_type);
 const NLAPolicy *nfnl_get_policy(uint16_t nlmsg_type);
+const NLAPolicy *sdnl_get_policy(uint16_t nlmsg_type, uint16_t nlmsg_flags);
 const NLAPolicySet *genl_get_policy_set_by_name(const char *name);
 int genl_get_policy_set_and_header_size(
                 sd_netlink *nl,
@@ -52,6 +53,7 @@ const NLAPolicySetUnion *policy_get_policy_set_union(const NLAPolicy *policy);
 int netlink_get_policy_set_and_header_size(
                 sd_netlink *nl,
                 uint16_t type,
+                uint16_t flags,
                 const NLAPolicySet **ret_policy_set,
                 size_t *ret_header_size);
 

src/libsystemd/sd-netlink/test-netlink.c

@@ -9,15 +9,19 @@
 #include <linux/if_macsec.h>
 #include <linux/l2tp.h>
 #include <linux/nl80211.h>
+#include <linux/unix_diag.h>
+#include <sys/stat.h>
 #include <unistd.h>
 
 #include "sd-netlink.h"
 
 #include "alloc-util.h"
 #include "ether-addr-util.h"
+#include "fd-util.h"
 #include "macro.h"
 #include "netlink-genl.h"
 #include "netlink-internal.h"
+#include "netlink-sock-diag.h"
 #include "netlink-util.h"
 #include "socket-util.h"
 #include "stdio-util.h"
@@ -700,4 +704,28 @@ TEST(rtnl_set_link_name) {
         ASSERT_NULL(resolved = mfree(resolved));
 }
 
+TEST(sock_diag_unix) {
+        _cleanup_(sd_netlink_unrefp) sd_netlink *nl = NULL;
+
+        ASSERT_OK(sd_sock_diag_socket_open(&nl));
+
+        _cleanup_close_ int unix_fd = ASSERT_FD(socket(AF_UNIX, SOCK_STREAM|SOCK_CLOEXEC, 0));
+        ASSERT_OK(socket_autobind(unix_fd, /* ret_name= */ NULL));
+        ASSERT_OK_ERRNO(listen(unix_fd, 123));
+
+        struct stat st;
+        ASSERT_OK_ERRNO(fstat(unix_fd, &st));
+
+        uint64_t cookie;
+        socklen_t cookie_len = sizeof(cookie);
+        ASSERT_OK_ERRNO(getsockopt(unix_fd, SOL_SOCKET, SO_COOKIE, &cookie, &cookie_len));
+        ASSERT_EQ(cookie_len, sizeof(cookie));
+
+        _cleanup_(sd_netlink_message_unrefp) sd_netlink_message *message = NULL;
+        ASSERT_OK(sd_sock_diag_message_new_unix(nl, &message, st.st_ino, cookie, UDIAG_SHOW_RQLEN));
+
+        _cleanup_(sd_netlink_message_unrefp) sd_netlink_message *reply = NULL;
+        ASSERT_OK(sd_netlink_call(nl, message, /* usec= */ 0, &reply));
+}
+
 DEFINE_TEST_MAIN(LOG_DEBUG);

src/login/logind-session-dbus.c

@@ -536,7 +536,7 @@ static int method_set_tty(sd_bus_message *message, void *userdata, sd_bus_error
         flags = fcntl(fd, F_GETFL, 0);
         if (flags < 0)
                 return -errno;
-        if ((flags & O_ACCMODE) != O_RDWR)
+        if ((flags & O_ACCMODE_STRICT) != O_RDWR)
                 return -EACCES;
         if (FLAGS_SET(flags, O_PATH))
                 return -ENOTTY;

src/mountfsd/mountwork.c

@@ -99,7 +99,7 @@ static int validate_image_fd(int fd, MountImageParameters *p) {
         if (fl < 0)
                 return log_debug_errno(fl, "Image file descriptor has unsafe flags set: %m");
 
-        switch (fl & O_ACCMODE) {
+        switch (fl & O_ACCMODE_STRICT) {
 
         case O_RDONLY:
                 p->read_only = true;

src/run/run.c

@@ -28,6 +28,8 @@
 #include "exec-util.h"
 #include "exit-status.h"
 #include "fd-util.h"
+#include "fork-journal.h"
+#include "format-table.h"
 #include "format-util.h"
 #include "fs-util.h"
 #include "hostname-setup.h"
@@ -89,6 +91,7 @@ static char **arg_socket_property = NULL;
 static char **arg_timer_property = NULL;
 static bool arg_with_timer = false;
 static bool arg_quiet = false;
+static bool arg_verbose = false;
 static bool arg_aggressive_gc = false;
 static char *arg_working_directory = NULL;
 static bool arg_shell = false;
@@ -155,6 +158,7 @@ static int help(void) {
                "                                  agents until unit is started up\n"
                "  -P --pipe                       Pass STDIN/STDOUT/STDERR directly to service\n"
                "  -q --quiet                      Suppress information messages during runtime\n"
+               "  -v --verbose                    Show unit logs while executing operation\n"
                "     --json=pretty|short|off      Print unit name and invocation id as JSON\n"
                "  -G --collect                    Unload unit after it ran, even when failed\n"
                "  -S --shell                      Invoke a $SHELL interactively\n"
@@ -332,6 +336,7 @@ static int parse_argv(int argc, char *argv[]) {
                 { "pty-late",           no_argument,       NULL, 'T'                    },
                 { "pipe",               no_argument,       NULL, 'P'                    },
                 { "quiet",              no_argument,       NULL, 'q'                    },
+                { "verbose",            no_argument,       NULL, 'v'                    },
                 { "on-active",          required_argument, NULL, ARG_ON_ACTIVE          },
                 { "on-boot",            required_argument, NULL, ARG_ON_BOOT            },
                 { "on-startup",         required_argument, NULL, ARG_ON_STARTUP         },
@@ -365,7 +370,7 @@ static int parse_argv(int argc, char *argv[]) {
         /* Resetting to 0 forces the invocation of an internal initialization routine of getopt_long()
          * that checks for GNU extensions in optstring ('-' or '+' at the beginning). */
         optind = 0;
-        while ((c = getopt_long(argc, argv, "+hrC:H:M:E:p:tTPqGdSu:", options, NULL)) >= 0)
+        while ((c = getopt_long(argc, argv, "+hrC:H:M:E:p:tTPqvGdSu:", options, NULL)) >= 0)
 
                 switch (c) {
 
@@ -492,6 +497,10 @@ static int parse_argv(int argc, char *argv[]) {
                         arg_quiet = true;
                         break;
 
+                case 'v':
+                        arg_verbose = true;
+                        break;
+
                 case ARG_ON_ACTIVE:
                         r = add_timer_property("OnActiveSec", optarg);
                         if (r < 0)
@@ -2137,6 +2146,142 @@ static int run_context_setup_ptyfwd(RunContext *c) {
         return 0;
 }
 
+static int run_context_show_result(RunContext *c) {
+        int r;
+
+        assert(c);
+
+        _cleanup_(table_unrefp) Table *t = table_new_vertical();
+        if (!t)
+                return log_oom();
+
+        if (!isempty(c->result)) {
+                r = table_add_many(
+                                t,
+                                TABLE_FIELD, "Finished with result",
+                                TABLE_STRING, c->result,
+                                TABLE_SET_COLOR, streq(c->result, "success") ? ansi_highlight_green() : ansi_highlight_red());
+                if (r < 0)
+                        return table_log_add_error(r);
+        }
+
+        if (c->exit_code > 0) {
+                r = table_add_cell(
+                                t,
+                                /* ret_cell= */ NULL,
+                                TABLE_FIELD,
+                                "Main processes terminated with");
+                if (r < 0)
+                        return table_log_add_error(r);
+
+                r = table_add_cell_stringf(
+                                t,
+                                /* ret_cell= */ NULL,
+                                "code=%s, status=%u/%s",
+                                sigchld_code_to_string(c->exit_code),
+                                c->exit_status,
+                                strna(c->exit_code == CLD_EXITED ?
+                                      exit_status_to_string(c->exit_status, EXIT_STATUS_FULL) :
+                                      signal_to_string(c->exit_status)));
+                if (r < 0)
+                        return table_log_add_error(r);
+        }
+
+        if (timestamp_is_set(c->inactive_enter_usec) &&
+            timestamp_is_set(c->inactive_exit_usec) &&
+            c->inactive_enter_usec > c->inactive_exit_usec) {
+                r = table_add_many(
+                                t,
+                                TABLE_FIELD, "Service runtime",
+                                TABLE_TIMESPAN_MSEC, c->inactive_enter_usec - c->inactive_exit_usec);
+                if (r < 0)
+                        return table_log_add_error(r);
+        }
+
+        if (c->cpu_usage_nsec != NSEC_INFINITY) {
+                r = table_add_many(
+                                t,
+                                TABLE_FIELD, "CPU time consumed",
+                                TABLE_TIMESPAN_MSEC, DIV_ROUND_UP(c->cpu_usage_nsec, NSEC_PER_USEC));
+                if (r < 0)
+                        return table_log_add_error(r);
+        }
+
+        if (c->memory_peak != UINT64_MAX) {
+                const char *swap;
+
+                if (c->memory_swap_peak != UINT64_MAX)
+                        swap = strjoina(" (swap: ", FORMAT_BYTES(c->memory_swap_peak), ")");
+                else
+                        swap = "";
+
+                r = table_add_cell(
+                                t,
+                                /* ret_cell= */ NULL,
+                                TABLE_FIELD, "Memory peak");
+                if (r < 0)
+                        return table_log_add_error(r);
+
+                r = table_add_cell_stringf(
+                                t,
+                                /* ret_cell= */ NULL,
+                                "%s%s",
+                                FORMAT_BYTES(c->memory_peak), swap);
+                if (r < 0)
+                        return table_log_add_error(r);
+        }
+
+        const char *ip_ingress = NULL, *ip_egress = NULL;
+        if (!IN_SET(c->ip_ingress_bytes, 0, UINT64_MAX))
+                ip_ingress = strjoina("received ", FORMAT_BYTES(c->ip_ingress_bytes));
+        if (!IN_SET(c->ip_egress_bytes, 0, UINT64_MAX))
+                ip_egress = strjoina("sent ", FORMAT_BYTES(c->ip_egress_bytes));
+
+        if (ip_ingress || ip_egress) {
+                r = table_add_cell(
+                                t,
+                                /* ret_cell= */ NULL,
+                                TABLE_FIELD, "IP Traffic");
+                if (r < 0)
+                        return table_log_add_error(r);
+
+                r = table_add_cell_stringf(
+                                t,
+                                /* ret_cell= */ NULL,
+                                "%s%s%s", strempty(ip_ingress), ip_ingress && ip_egress ? ", " : "", strempty(ip_egress));
+                if (r < 0)
+                        return table_log_add_error(r);
+        }
+
+        const char *io_read = NULL, *io_write = NULL;
+        if (!IN_SET(c->io_read_bytes, 0, UINT64_MAX))
+                io_read = strjoina("read ", FORMAT_BYTES(c->io_read_bytes));
+        if (!IN_SET(c->io_write_bytes, 0, UINT64_MAX))
+                io_write = strjoina("written ", FORMAT_BYTES(c->io_write_bytes));
+
+        if (io_read || io_write) {
+                r = table_add_cell(
+                                t,
+                                /* ret_cell= */ NULL,
+                                TABLE_FIELD, "IO Bytes");
+                if (r < 0)
+                        return table_log_add_error(r);
+
+                r = table_add_cell_stringf(
+                                t,
+                                /* ret_cell= */ NULL,
+                                "%s%s%s", strempty(io_read), io_read && io_write ? ", " : "", strempty(io_write));
+                if (r < 0)
+                        return table_log_add_error(r);
+        }
+
+        r = table_print(t, stderr);
+        if (r < 0)
+                return table_log_print_error(r);
+
+        return 0;
+}
+
 static int start_transient_service(sd_bus *bus) {
         _cleanup_(sd_bus_message_unrefp) sd_bus_message *m = NULL, *reply = NULL;
         _cleanup_(sd_bus_error_free) sd_bus_error error = SD_BUS_ERROR_NULL;
@@ -2251,6 +2396,10 @@ static int start_transient_service(sd_bus *bus) {
                 return r;
         peer_fd = safe_close(peer_fd);
 
+        _cleanup_(journal_terminate) PidRef journal_pid = PIDREF_NULL;
+        if (arg_verbose)
+                (void) journal_fork(arg_runtime_scope, STRV_MAKE(c.unit), &journal_pid);
+
         r = bus_call_with_hint(bus, m, "service", &reply);
         if (r < 0)
                 return r;
@@ -2325,60 +2474,11 @@ static int start_transient_service(sd_bus *bus) {
                 if (r < 0)
                         return log_error_errno(r, "Failed to run event loop: %m");
 
-                if (arg_wait && !arg_quiet) {
+                /* Close the journal watch logic before we output the exit summary */
+                journal_terminate(&journal_pid);
 
-                        if (!isempty(c.result))
-                                log_info("Finished with result: %s", strna(c.result));
-
-                        if (c.exit_code > 0)
-                                log_info("Main processes terminated with: code=%s, status=%u/%s",
-                                         sigchld_code_to_string(c.exit_code),
-                                         c.exit_status,
-                                         strna(c.exit_code == CLD_EXITED ?
-                                               exit_status_to_string(c.exit_status, EXIT_STATUS_FULL) :
-                                               signal_to_string(c.exit_status)));
-
-                        if (timestamp_is_set(c.inactive_enter_usec) &&
-                            timestamp_is_set(c.inactive_exit_usec) &&
-                            c.inactive_enter_usec > c.inactive_exit_usec)
-                                log_info("Service runtime: %s",
-                                         FORMAT_TIMESPAN(c.inactive_enter_usec - c.inactive_exit_usec, USEC_PER_MSEC));
-
-                        if (c.cpu_usage_nsec != NSEC_INFINITY)
-                                log_info("CPU time consumed: %s",
-                                         FORMAT_TIMESPAN(DIV_ROUND_UP(c.cpu_usage_nsec, NSEC_PER_USEC), USEC_PER_MSEC));
-
-                        if (c.memory_peak != UINT64_MAX) {
-                                const char *swap;
-
-                                if (c.memory_swap_peak != UINT64_MAX)
-                                        swap = strjoina(" (swap: ", FORMAT_BYTES(c.memory_swap_peak), ")");
-                                else
-                                        swap = "";
-
-                                log_info("Memory peak: %s%s", FORMAT_BYTES(c.memory_peak), swap);
-                        }
-
-                        const char *ip_ingress = NULL, *ip_egress = NULL;
-
-                        if (!IN_SET(c.ip_ingress_bytes, 0, UINT64_MAX))
-                                ip_ingress = strjoina(" received: ", FORMAT_BYTES(c.ip_ingress_bytes));
-                        if (!IN_SET(c.ip_egress_bytes, 0, UINT64_MAX))
-                                ip_egress = strjoina(" sent: ", FORMAT_BYTES(c.ip_egress_bytes));
-
-                        if (ip_ingress || ip_egress)
-                                log_info("IP traffic%s%s", strempty(ip_ingress), strempty(ip_egress));
-
-                        const char *io_read = NULL, *io_write = NULL;
-
-                        if (!IN_SET(c.io_read_bytes, 0, UINT64_MAX))
-                                io_read = strjoina(" read: ", FORMAT_BYTES(c.io_read_bytes));
-                        if (!IN_SET(c.io_write_bytes, 0, UINT64_MAX))
-                                io_write = strjoina(" written: ", FORMAT_BYTES(c.io_write_bytes));
-
-                        if (io_read || io_write)
-                                log_info("IO bytes%s%s", strempty(io_read), strempty(io_write));
-                }
+                if (arg_wait && !arg_quiet)
+                        run_context_show_result(&c);
 
                 /* Try to propagate the service's return value. But if the service defines
                  * e.g. SuccessExitStatus, honour this, and return 0 to mean "success". */

src/shared/data-fd-util.c

@@ -146,13 +146,13 @@ int memfd_clone_fd(int fd, const char *name, int mode) {
 
         assert(fd >= 0);
         assert(name);
-        assert(IN_SET(mode & O_ACCMODE, O_RDONLY, O_RDWR));
+        assert(IN_SET(mode & O_ACCMODE_STRICT, O_RDONLY, O_RDWR));
         assert((mode & ~(O_RDONLY|O_RDWR|O_CLOEXEC)) == 0);
 
         if (fstat(fd, &st) < 0)
                 return -errno;
 
-        ro = (mode & O_ACCMODE) == O_RDONLY;
+        ro = (mode & O_ACCMODE_STRICT) == O_RDONLY;
         exec = st.st_mode & 0111;
 
         mfd = memfd_create_wrapper(name,

src/shared/fork-journal.c

@@ -0,0 +1,216 @@
+/* SPDX-License-Identifier: LGPL-2.1-or-later */
+
+#include <sys/types.h>
+#include <unistd.h>
+
+#include "build-path.h"
+#include "escape.h"
+#include "event-util.h"
+#include "exit-status.h"
+#include "fd-util.h"
+#include "fork-journal.h"
+#include "notify-recv.h"
+#include "parse-util.h"
+#include "process-util.h"
+#include "signal-util.h"
+#include "socket-util.h"
+#include "strv.h"
+
+static int on_child_exit(sd_event_source *s, const siginfo_t *si, void *userdata) {
+        PidRef *child = ASSERT_PTR(userdata);
+
+        assert(si->si_pid == child->pid);
+
+        /* Let's first do some debug logging about the exit status of the child */
+
+        if (si->si_code == CLD_EXITED) {
+                if (si->si_status == EXIT_SUCCESS)
+                        log_debug("journalctl " PID_FMT " exited successfully.", si->si_pid);
+                else
+                        log_debug("journalctl " PID_FMT " died with a failure exit status %i, ignoring.", si->si_pid, si->si_status);
+        } else if (si->si_code == CLD_KILLED)
+                log_debug("journalctl " PID_FMT " was killed by signal %s, ignoring.", si->si_pid, signal_to_string(si->si_status));
+        else if (si->si_code == CLD_DUMPED)
+                log_debug("journalctl " PID_FMT " dumped core by signal %s, ignoring.", si->si_pid, signal_to_string(si->si_status));
+        else
+                log_debug("Got unexpected exit code via SIGCHLD, ignoring.");
+
+        /* And let's then fail the whole thing, because regardless what the exit status of the child is
+         * (i.e. even if successful), if it exits before sending READY=1 something is wrong. */
+
+        return log_debug_errno(SYNTHETIC_ERRNO(EPROTO), "Child " PID_FMT " died before sending notification message.", child->pid);
+}
+
+static int on_child_notify(sd_event_source *s, int fd, uint32_t revents, void *userdata) {
+        PidRef *child = ASSERT_PTR(userdata);
+        int r;
+
+        _cleanup_strv_free_ char **msg = NULL;
+        _cleanup_(pidref_done) PidRef sender = PIDREF_NULL;
+        r = notify_recv_strv(fd, &msg, /* ret_ucred= */ NULL, &sender);
+        if (r == -EAGAIN)
+                return 0;
+        if (r < 0)
+                return r;
+
+        if (!pidref_equal(child, &sender)) {
+                log_warning("Received notification message from unexpected process " PID_FMT " (expected " PID_FMT "), ignoring.",
+                            sender.pid, child->pid);
+                return 0;
+        }
+
+        if (strv_find(msg, "READY=1"))
+                return sd_event_exit(sd_event_source_get_event(s), EXIT_SUCCESS);
+
+        const char *e = strv_find_startswith(msg, "ERRNO=");
+        if (e) {
+                int error;
+
+                r = safe_atoi(e, &error);
+                if (r < 0) {
+                        log_debug_errno(r, "Received invalid ERRNO= notification message, ignoring: %m");
+                        return 0;
+                }
+                if (error <= 0) {
+                        log_debug("Received non-positive ERRNO= notification message, ignoring: %m");
+                        return 0;
+                }
+
+                return -error;
+        }
+
+        return 0;
+}
+
+int journal_fork(RuntimeScope scope, char **units, PidRef *ret_pidref) {
+        int r;
+
+        assert(scope >= 0);
+        assert(scope < _RUNTIME_SCOPE_MAX);
+        assert(ret_pidref);
+
+        if (!is_main_thread())
+                return -EPERM;
+
+        if (strv_isempty(units))
+                return 0;
+
+        _cleanup_close_ int notify_fd = socket(AF_UNIX, SOCK_DGRAM|SOCK_CLOEXEC|SOCK_NONBLOCK, 0);
+        if (notify_fd < 0)
+                return log_debug_errno(errno, "Failed to allocate AF_UNIX socket for notifications: %m");
+
+        r = setsockopt_int(notify_fd, SOL_SOCKET, SO_PASSCRED, true);
+        if (r < 0)
+                return log_debug_errno(r, "Failed to enable SO_PASSCRED: %m");
+
+        r = setsockopt_int(notify_fd, SOL_SOCKET, SO_PASSPIDFD, true);
+        if (r < 0)
+                log_debug_errno(r, "Failed to enable SO_PASSPIDFD, ignoring: %m");
+
+        /* Pick an address via auto-bind */
+        _cleanup_free_ char *addr_string = NULL;
+        r = socket_autobind(notify_fd, &addr_string);
+        if (r < 0)
+                return log_debug_errno(r, "Failed to bind AF_UNIX socket: %m");
+
+        _cleanup_strv_free_ char **argv = strv_new(
+                        "journalctl",
+                        "-q",
+                        "--follow",
+                        "--no-pager",
+                        "--lines=1",
+                        "--synchronize-on-exit=yes");
+        if (!argv)
+                return log_oom_debug();
+
+        STRV_FOREACH(u, units)
+                if (strv_extendf(&argv,
+                                 scope == RUNTIME_SCOPE_SYSTEM ? "--unit=%s" : "--user-unit=%s",
+                                 *u) < 0)
+                        return log_oom_debug();
+
+        if (DEBUG_LOGGING) {
+                _cleanup_free_ char *l = quote_command_line(argv, SHELL_ESCAPE_EMPTY);
+                log_debug("Invoking '%s' as child.", strnull(l));
+        }
+
+        BLOCK_SIGNALS(SIGCHLD);
+
+        _cleanup_(pidref_done_sigkill_wait) PidRef child = PIDREF_NULL;
+        r = pidref_safe_fork_full(
+                        "(journalctl)",
+                        (const int[3]) { -EBADF, STDOUT_FILENO, STDERR_FILENO },
+                        /* except_fds= */ NULL,
+                        /* n_except_fds= */ 0,
+                        FORK_RESET_SIGNALS|FORK_DEATHSIG_SIGTERM|FORK_CLOSE_ALL_FDS|FORK_RLIMIT_NOFILE_SAFE|FORK_REARRANGE_STDIO,
+                        &child);
+        if (r < 0)
+                return r;
+        if (r == 0) {
+                /* In the child: */
+
+                if (setenv("NOTIFY_SOCKET", addr_string, /* overwrite= */ true) < 0) {
+                        log_debug_errno(errno, "Failed to set $NOTIFY_SOCKET: %m");
+                        _exit(EXIT_MEMORY);
+                }
+
+                r = invoke_callout_binary(argv[0], (char**) argv);
+                log_debug_errno(r, "Failed to invoke journalctl: %m");
+                _exit(EXIT_EXEC);
+        }
+
+        _cleanup_(sd_event_unrefp) sd_event *event = NULL;
+        r = sd_event_new(&event);
+        if (r < 0)
+                return r;
+
+        _cleanup_(sd_event_source_disable_unrefp) sd_event_source *child_event_source = NULL;
+        r = event_add_child_pidref(event, &child_event_source, &child, WEXITED, on_child_exit, &child);
+        if (r < 0)
+                return r;
+
+        r = sd_event_source_set_exit_on_failure(child_event_source, true);
+        if (r < 0)
+                return r;
+
+        (void) sd_event_source_set_description(child_event_source, "fork-journal-child");
+
+        _cleanup_(sd_event_source_disable_unrefp) sd_event_source *notify_event_source = NULL;
+        r = sd_event_add_io(event, &notify_event_source, notify_fd, EPOLLIN, on_child_notify, &child);
+        if (r < 0)
+                return r;
+
+        r = sd_event_source_set_exit_on_failure(notify_event_source, true);
+        if (r < 0)
+                return r;
+
+        /* We want the notification message from the child before the SIGCHLD */
+        r = sd_event_source_set_priority(notify_event_source, SD_EVENT_PRIORITY_NORMAL-10);
+        if (r < 0)
+                return r;
+
+        (void) sd_event_source_set_description(notify_event_source, "fork-journal-notify");
+
+        r = sd_event_loop(event);
+        if (r < 0)
+                return r;
+        assert(r == 0);
+
+        *ret_pidref = TAKE_PIDREF(child);
+
+        return 0;
+}
+
+void journal_terminate(PidRef *pidref) {
+        int r;
+
+        if (!pidref_is_set(pidref))
+                return;
+
+        r = pidref_kill(pidref, SIGTERM);
+        if (r < 0)
+                log_debug_errno(r, "Failed to send SIGTERM to journalctl child " PID_FMT ", ignoring: %m", pidref->pid);
+
+        (void) pidref_wait_for_terminate_and_check("journalctl", pidref, /* flags= */ 0);
+        pidref_done(pidref);
+}

src/shared/fork-journal.h

@@ -0,0 +1,10 @@
+/* SPDX-License-Identifier: LGPL-2.1-or-later */
+#pragma once
+
+#include "macro.h"
+#include "runtime-scope.h"
+#include "set.h"
+
+int journal_fork(RuntimeScope scope, char **units, PidRef *ret_pidref);
+
+void journal_terminate(PidRef *pidref);

src/shared/journal-file-util.c

@@ -504,7 +504,7 @@ int journal_file_open_reliably(
                     -EIDRM))            /* File has been deleted */
                 return r;
 
-        if ((open_flags & O_ACCMODE) == O_RDONLY)
+        if ((open_flags & O_ACCMODE_STRICT) == O_RDONLY)
                 return r;
 
         if (!(open_flags & O_CREAT))
@@ -519,7 +519,7 @@ int journal_file_open_reliably(
         /* The file is corrupted. Try opening it read-only as the template before rotating to inherit its
          * sequence number and ID. */
         r = journal_file_open(-EBADF, fname,
-                              (open_flags & ~(O_ACCMODE|O_CREAT|O_EXCL)) | O_RDONLY,
+                              (open_flags & ~(O_ACCMODE_STRICT|O_CREAT|O_EXCL)) | O_RDONLY,
                               file_flags, 0, compress_threshold_bytes, NULL,
                               mmap_cache, /* template = */ NULL, &old_file);
         if (r < 0)

src/shared/logs-show.c

@@ -363,7 +363,6 @@ finish:
 
 static int output_timestamp_realtime(
                 FILE *f,
-                sd_journal *j,
                 OutputMode mode,
                 OutputFlags flags,
                 usec_t usec) {
@@ -372,7 +371,6 @@ static int output_timestamp_realtime(
         int r;
 
         assert(f);
-        assert(j);
 
         if (!VALID_REALTIME(usec))
                 return log_debug_errno(SYNTHETIC_ERRNO(EINVAL), "No valid realtime timestamp available, skipping showing journal entry.");
@@ -621,7 +619,7 @@ static int output_short(
         } else {
                 usec_t usec;
                 parse_display_realtime(j, realtime, monotonic, &usec);
-                r = output_timestamp_realtime(f, j, mode, flags, usec);
+                r = output_timestamp_realtime(f, mode, flags, usec);
         }
         if (r == -EINVAL)
                 return 0;
@@ -813,7 +811,7 @@ static int output_verbose(
 
         r = get_display_realtime(j, &usec);
         if (IN_SET(r, -EBADMSG, -EADDRNOTAVAIL)) {
-                log_debug_errno(r, "Skipping message we can't read: %m");
+                log_debug_errno(r, "Unable to read realtime timestamp from entry, assuming bad or partially written entry: %m");
                 return 0;
         }
         if (r < 0)
@@ -823,6 +821,10 @@ static int output_verbose(
                 return log_error_errno(SYNTHETIC_ERRNO(EINVAL), "No valid realtime timestamp available");
 
         r = sd_journal_get_cursor(j, &cursor);
+        if (r == -EBADMSG) {
+                log_debug_errno(r, "Unable to determine cursor for entry, assuming bad or partially written entry: %m");
+                return 0;
+        }
         if (r < 0)
                 return log_error_errno(r, "Failed to get cursor: %m");
 
@@ -843,12 +845,16 @@ static int output_verbose(
                 size_t fieldlen, valuelen;
 
                 c = memchr(data, '=', length);
-                if (!c)
-                        return log_error_errno(SYNTHETIC_ERRNO(EINVAL), "Invalid field.");
+                if (!c) {
+                        log_debug("Encountered field without '=', assuming bad or partially written entry, leaving.");
+                        break;
+                }
 
                 fieldlen = c - (const char*) data;
-                if (!journal_field_valid(data, fieldlen, true))
-                        return log_error_errno(SYNTHETIC_ERRNO(EINVAL), "Invalid field.");
+                if (!journal_field_valid(data, fieldlen, /* allow_protected= */ true)) {
+                        log_debug("Encountered invalid field, assuming bad or partially written entry, leaving.");
+                        break;
+                }
 
                 r = field_set_test(output_fields, data, fieldlen);
                 if (r < 0)
@@ -860,10 +866,10 @@ static int output_verbose(
                 p = c + 1;
 
                 if (flags & OUTPUT_COLOR) {
-                        if (startswith(data, "MESSAGE=")) {
+                        if (memory_startswith(data, length, "MESSAGE=")) {
                                 on = ansi_highlight();
                                 off = ansi_normal();
-                        } else if (startswith(data, "CONFIG_FILE=")) {
+                        } else if (memory_startswith(data, length, "CONFIG_FILE=")) {
                                 _cleanup_free_ char *u = NULL;
 
                                 u = memdup_suffix0(p, valuelen);
@@ -875,7 +881,7 @@ static int output_verbose(
                                         valuelen = strlen(urlified);
                                 }
 
-                        } else if (startswith(data, "_")) {
+                        } else if (memory_startswith(data, length, "_")) {
                                 /* Highlight trusted data as such */
                                 on = ansi_green();
                                 off = ansi_normal();
@@ -931,18 +937,34 @@ static int output_export(
         (void) sd_journal_set_data_threshold(j, 0);
 
         r = sd_journal_get_cursor(j, &cursor);
+        if (IN_SET(r, -EBADMSG, -EADDRNOTAVAIL)) {
+                log_debug_errno(r, "Unable to determine cursor of entry, assuming bad or partially written entry: %m");
+                return 0;
+        }
         if (r < 0)
                 return log_error_errno(r, "Failed to get cursor: %m");
 
         r = sd_journal_get_realtime_usec(j, &realtime);
+        if (r == -EBADMSG) {
+                log_debug_errno(r, "Unable to read realtime timestamp of entry, assuming bad or partially written entry: %m");
+                return 0;
+        }
         if (r < 0)
                 return log_error_errno(r, "Failed to get realtime timestamp: %m");
 
         r = sd_journal_get_monotonic_usec(j, &monotonic, &journal_boot_id);
+        if (r == -EBADMSG) {
+                log_debug_errno(r, "Unable to read monotonic timestamp of entry, assuming bad or partially written entry: %m");
+                return 0;
+        }
         if (r < 0)
                 return log_error_errno(r, "Failed to get monotonic timestamp: %m");
 
         r = sd_journal_get_seqnum(j, &seqnum, &seqnum_id);
+        if (r == -EBADMSG) {
+                log_debug_errno(r, "Unable to read sequence number of entry, assuming bad or partially written entry: %m");
+                return 0;
+        }
         if (r < 0)
                 return log_error_errno(r, "Failed to get seqnum: %m");
 
@@ -969,12 +991,16 @@ static int output_export(
                         continue;
 
                 c = memchr(data, '=', length);
-                if (!c)
-                        return log_error_errno(SYNTHETIC_ERRNO(EINVAL), "Invalid field.");
+                if (!c) {
+                        log_debug("Encountered data field without '=', assuming bad or partially written entry, leaving.");
+                        break;
+                }
 
                 fieldlen = c - (const char*) data;
-                if (!journal_field_valid(data, fieldlen, true))
-                        return log_error_errno(SYNTHETIC_ERRNO(EINVAL), "Invalid field.");
+                if (!journal_field_valid(data, fieldlen, /* allow_protected= */ true)) {
+                        log_debug("Encountered invalid field, assuming bad or partially written entry, leaving.");
+                        break;
+                }
 
                 r = field_set_test(output_fields, data, fieldlen);
                 if (r < 0)
@@ -1160,8 +1186,10 @@ static int update_json_data_split(
                 return 0;
 
         fieldlen = eq - (const char*) data;
-        if (!journal_field_valid(data, fieldlen, true))
-                return log_error_errno(SYNTHETIC_ERRNO(EINVAL), "Invalid field.");
+        if (!journal_field_valid(data, fieldlen, /* allow_protected= */ true)) {
+                log_debug("Encountered invalid field, assuming bad or incompletely written field, leaving.");
+                return 0;
+        }
 
         name = strndupa_safe(data, fieldlen);
         if (output_fields && !set_contains(output_fields, name))
@@ -1198,18 +1226,34 @@ static int output_json(
         (void) sd_journal_set_data_threshold(j, flags & OUTPUT_SHOW_ALL ? 0 : JSON_THRESHOLD);
 
         r = sd_journal_get_cursor(j, &cursor);
+        if (IN_SET(r, -EBADMSG, -EADDRNOTAVAIL)) {
+                log_debug_errno(r, "Unable to determine cursor of entry, assuming bad or partially written entry: %m");
+                return 0;
+        }
         if (r < 0)
                 return log_error_errno(r, "Failed to get cursor: %m");
 
         r = sd_journal_get_realtime_usec(j, &realtime);
+        if (r == -EBADMSG) {
+                log_debug_errno(r, "Unable to read realtime timestamp of entry, assuming bad or partially written entry: %m");
+                return 0;
+        }
         if (r < 0)
                 return log_error_errno(r, "Failed to get realtime timestamp: %m");
 
         r = sd_journal_get_monotonic_usec(j, &monotonic, &journal_boot_id);
+        if (r == -EBADMSG) {
+                log_debug_errno(r, "Unable to read monotonic timestamp of entry, assuming bad or partially written entry: %m");
+                return 0;
+        }
         if (r < 0)
                 return log_error_errno(r, "Failed to get monotonic timestamp: %m");
 
         r = sd_journal_get_seqnum(j, &seqnum, &seqnum_id);
+        if (r == -EBADMSG) {
+                log_debug_errno(r, "Unable to read sequence number of entry, assuming bad or partially written entry: %m");
+                return 0;
+        }
         if (r < 0)
                 return log_error_errno(r, "Failed to get seqnum: %m");
 
@@ -1525,6 +1569,10 @@ int show_journal(
 
                 if (need_seek) {
                         r = sd_journal_next(j);
+                        if (r == -EBADMSG) {
+                                log_debug_errno(r, "Bad or partially written entry, leaving.");
+                                break;
+                        }
                         if (r < 0)
                                 return log_error_errno(r, "Failed to iterate through journal: %m");
                 }

src/shared/loop-util.c

@@ -500,7 +500,7 @@ static int loop_device_make_internal(
                 .block_size = sector_size,
                 .info = {
                         /* Use the specified flags, but configure the read-only flag from the open flags, and force autoclear */
-                        .lo_flags = (loop_flags & ~LO_FLAGS_READ_ONLY) | ((open_flags & O_ACCMODE) == O_RDONLY ? LO_FLAGS_READ_ONLY : 0) | LO_FLAGS_AUTOCLEAR,
+                        .lo_flags = (loop_flags & ~LO_FLAGS_READ_ONLY) | ((open_flags & O_ACCMODE_STRICT) == O_RDONLY ? LO_FLAGS_READ_ONLY : 0) | LO_FLAGS_AUTOCLEAR,
                         .lo_offset = offset,
                         .lo_sizelimit = size == UINT64_MAX ? 0 : size,
                 },

src/shared/meson.build

@@ -76,6 +76,7 @@ shared_sources = files(
         'find-esp.c',
         'firewall-util-nft.c',
         'firewall-util.c',
+        'fork-journal.c',
         'format-table.c',
         'fstab-util.c',
         'generator.c',

src/shared/socket-netlink.c

@@ -5,7 +5,11 @@
 #include <arpa/inet.h>
 #include <errno.h>
 #include <linux/net_namespace.h>
+#include <linux/unix_diag.h>
 #include <string.h>
+#include <sys/stat.h>
+
+#include "sd-netlink.h"
 
 #include "alloc-util.h"
 #include "errno-util.h"
@@ -14,6 +18,7 @@
 #include "log.h"
 #include "memory-util.h"
 #include "namespace-util.h"
+#include "netlink-sock-diag.h"
 #include "netlink-util.h"
 #include "parse-util.h"
 #include "socket-netlink.h"
@@ -480,3 +485,63 @@ int netns_get_nsid(int netnsfd, uint32_t *ret) {
 
         return -ENXIO;
 }
+
+int af_unix_get_qlen(int fd, uint32_t *ret) {
+        int r;
+
+        assert(fd >= 0);
+        assert(ret);
+
+        /* Returns the current queue length for an AF_UNIX listening socket */
+
+        struct stat st;
+        if (fstat(fd, &st) < 0)
+                return -errno;
+        if (!S_ISSOCK(st.st_mode))
+                return -ENOTSOCK;
+
+        _cleanup_(sd_netlink_unrefp) sd_netlink *nl = NULL;
+        r = sd_sock_diag_socket_open(&nl);
+        if (r < 0)
+                return r;
+
+        uint64_t cookie = 0;
+        socklen_t cookie_len = sizeof(cookie);
+        if (getsockopt(fd, SOL_SOCKET, SO_COOKIE, &cookie, &cookie_len) < 0)
+                return -errno;
+
+        assert(cookie_len == sizeof(cookie));
+
+        _cleanup_(sd_netlink_message_unrefp) sd_netlink_message *message = NULL;
+        r = sd_sock_diag_message_new_unix(nl, &message, st.st_ino, cookie, UDIAG_SHOW_RQLEN);
+        if (r < 0)
+                return r;
+
+        _cleanup_(sd_netlink_message_unrefp) sd_netlink_message *reply = NULL;
+        r = sd_netlink_call(nl, message, /* usec= */ 0, &reply);
+        if (r < 0)
+                return r;
+
+        for (sd_netlink_message *m = reply; m; m = sd_netlink_message_next(m)) {
+                r = sd_netlink_message_get_errno(m);
+                if (r < 0)
+                        return r;
+
+                _cleanup_free_ void *data = NULL;
+                size_t size = 0;
+
+                r = sd_netlink_message_read_data(m, UNIX_DIAG_RQLEN, &size, &data);
+                if (r == -ENODATA)
+                        continue;
+                if (r < 0)
+                        return r;
+
+                assert(size == sizeof(struct unix_diag_rqlen));
+                const struct unix_diag_rqlen *udrql = ASSERT_PTR(data);
+
+                *ret = udrql->udiag_rqueue;
+                return 0;
+        }
+
+        return -ENODATA;
+}

src/shared/socket-netlink.h

@@ -45,3 +45,5 @@ int in_addr_full_new_from_string(const char *s, struct in_addr_full **ret);
 const char* in_addr_full_to_string(struct in_addr_full *a);
 
 int netns_get_nsid(int netnsfd, uint32_t *ret);
+
+int af_unix_get_qlen(int fd, uint32_t *ret);

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

@@ -2,7 +2,11 @@
 
 #include "varlink-io.systemd.Journal.h"
 
-static SD_VARLINK_DEFINE_METHOD(Synchronize);
+static SD_VARLINK_DEFINE_METHOD(
+                Synchronize,
+                SD_VARLINK_FIELD_COMMENT("Controls whether to offline the journal files as part of the synchronization operation."),
+                SD_VARLINK_DEFINE_INPUT(offline, SD_VARLINK_BOOL, SD_VARLINK_NULLABLE));
+
 static SD_VARLINK_DEFINE_METHOD(Rotate);
 static SD_VARLINK_DEFINE_METHOD(FlushToVar);
 static SD_VARLINK_DEFINE_METHOD(RelinquishVar);

src/systemctl/systemctl-start-unit.c

@@ -9,6 +9,7 @@
 #include "bus-util.h"
 #include "bus-wait-for-jobs.h"
 #include "bus-wait-for-units.h"
+#include "fork-journal.h"
 #include "macro.h"
 #include "special.h"
 #include "string-util.h"
@@ -387,9 +388,13 @@ int verb_start(int argc, char *argv[], void *userdata) {
                         return log_error_errno(r, "Failed to allocate unit watch context: %m");
         }
 
+        _cleanup_(journal_terminate) PidRef journal_pid = PIDREF_NULL;
         if (arg_marked)
                 ret = enqueue_marked_jobs(bus, w);
-        else
+        else {
+                if (arg_verbose)
+                        (void) journal_fork(arg_runtime_scope, names, &journal_pid);
+
                 STRV_FOREACH(name, names) {
                         _cleanup_(sd_bus_error_free) sd_bus_error error = SD_BUS_ERROR_NULL;
 
@@ -403,6 +408,7 @@ int verb_start(int argc, char *argv[], void *userdata) {
                                         return log_oom();
                         }
                 }
+        }
 
         if (!arg_no_block) {
                 const char *extra_args[4];

src/systemctl/systemctl.c

@@ -88,6 +88,7 @@ bool arg_show_types = false;
 int arg_check_inhibitors = -1;
 bool arg_dry_run = false;
 bool arg_quiet = false;
+bool arg_verbose = false;
 bool arg_no_warn = false;
 bool arg_full = false;
 bool arg_recursive = false;
@@ -302,6 +303,7 @@ static int systemctl_help(void) {
                "                             suspend-then-hibernate, hybrid-sleep, default,\n"
                "                             rescue, emergency, and exit.\n"
                "  -q --quiet             Suppress output\n"
+               "  -v --verbose           Show unit logs while executing operation\n"
                "     --no-warn           Suppress several warnings shown by default\n"
                "     --wait              For (re)start, wait until service stopped again\n"
                "                         For is-system-running, wait until startup is completed\n"
@@ -508,6 +510,7 @@ static int systemctl_parse_argv(int argc, char *argv[]) {
                 { "no-wall",             no_argument,       NULL, ARG_NO_WALL             },
                 { "dry-run",             no_argument,       NULL, ARG_DRY_RUN             },
                 { "quiet",               no_argument,       NULL, 'q'                     },
+                { "verbose",             no_argument,       NULL, 'v'                     },
                 { "no-warn",             no_argument,       NULL, ARG_NO_WARN             },
                 { "root",                required_argument, NULL, ARG_ROOT                },
                 { "image",               required_argument, NULL, ARG_IMAGE               },
@@ -554,7 +557,7 @@ static int systemctl_parse_argv(int argc, char *argv[]) {
         /* We default to allowing interactive authorization only in systemctl (not in the legacy commands) */
         arg_ask_password = true;
 
-        while ((c = getopt_long(argc, argv, "hC:t:p:P:alqfs:H:M:n:o:iTr.::", options, NULL)) >= 0)
+        while ((c = getopt_long(argc, argv, "hC:t:p:P:alqvfs:H:M:n:o:iTr.::", options, NULL)) >= 0)
 
                 switch (c) {
 
@@ -768,6 +771,10 @@ static int systemctl_parse_argv(int argc, char *argv[]) {
 
                         break;
 
+                case 'v':
+                        arg_verbose = true;
+                        break;
+
                 case 'f':
                         arg_force++;
                         break;

src/systemctl/systemctl.h

@@ -68,6 +68,7 @@ extern bool arg_show_types;
 extern int arg_check_inhibitors;
 extern bool arg_dry_run;
 extern bool arg_quiet;
+extern bool arg_verbose;
 extern bool arg_no_warn;
 extern bool arg_full;
 extern bool arg_recursive;

src/test/test-fileio.c

@@ -1104,7 +1104,7 @@ TEST(fdopen_independent) {
         zero(buf);
         assert_se(fread(buf, 1, sizeof(buf), f) == strlen(TEST_TEXT));
         ASSERT_STREQ(buf, TEST_TEXT);
-        assert_se((fcntl(fileno(f), F_GETFL) & O_ACCMODE) == O_RDONLY);
+        assert_se((fcntl(fileno(f), F_GETFL) & O_ACCMODE_STRICT) == O_RDONLY);
         assert_se(FLAGS_SET(fcntl(fileno(f), F_GETFD), FD_CLOEXEC));
         f = safe_fclose(f);
 
@@ -1112,7 +1112,7 @@ TEST(fdopen_independent) {
         zero(buf);
         assert_se(fread(buf, 1, sizeof(buf), f) == strlen(TEST_TEXT));
         ASSERT_STREQ(buf, TEST_TEXT);
-        assert_se((fcntl(fileno(f), F_GETFL) & O_ACCMODE) == O_RDONLY);
+        assert_se((fcntl(fileno(f), F_GETFL) & O_ACCMODE_STRICT) == O_RDONLY);
         assert_se(!FLAGS_SET(fcntl(fileno(f), F_GETFD), FD_CLOEXEC));
         f = safe_fclose(f);
 
@@ -1120,7 +1120,7 @@ TEST(fdopen_independent) {
         zero(buf);
         assert_se(fread(buf, 1, sizeof(buf), f) == strlen(TEST_TEXT));
         ASSERT_STREQ(buf, TEST_TEXT);
-        assert_se((fcntl(fileno(f), F_GETFL) & O_ACCMODE) == O_RDWR);
+        assert_se((fcntl(fileno(f), F_GETFL) & O_ACCMODE_STRICT) == O_RDWR);
         assert_se(FLAGS_SET(fcntl(fileno(f), F_GETFD), FD_CLOEXEC));
         f = safe_fclose(f);
 }

src/test/test-socket-netlink.c

@@ -1,10 +1,13 @@
 /* SPDX-License-Identifier: LGPL-2.1-or-later */
 
+#include <sys/eventfd.h>
+
 #include "alloc-util.h"
+#include "fd-util.h"
 #include "missing_network.h"
-#include "tests.h"
 #include "socket-netlink.h"
 #include "string-util.h"
+#include "tests.h"
 
 static void test_socket_address_parse_one(const char *in, int ret, int family, const char *expected) {
         SocketAddress a;
@@ -381,4 +384,33 @@ TEST(netns_get_nsid) {
                 log_info("Our NSID is %" PRIu32, u);
 }
 
+TEST(af_unix_get_qlen) {
+
+        _cleanup_close_ int unix_fd = ASSERT_FD(socket(AF_UNIX, SOCK_STREAM|SOCK_CLOEXEC, 0));
+        ASSERT_OK(socket_autobind(unix_fd, /* ret_name= */ NULL));
+        ASSERT_OK_ERRNO(listen(unix_fd, 123));
+
+        uint32_t q;
+        ASSERT_OK(af_unix_get_qlen(unix_fd, &q));
+        ASSERT_EQ(q, 0U);
+
+        _cleanup_close_ int conn_fd = ASSERT_FD(socket(AF_UNIX, SOCK_STREAM|SOCK_CLOEXEC|SOCK_NONBLOCK, 0));
+        union sockaddr_union sa;
+        socklen_t salen = sizeof(sa);
+        ASSERT_OK_ERRNO(getsockname(unix_fd, &sa.sa, &salen));
+        ASSERT_OK(connect(conn_fd, &sa.sa, salen));
+
+        ASSERT_OK(af_unix_get_qlen(unix_fd, &q));
+        ASSERT_EQ(q, 1U);
+
+        _cleanup_close_ int conn2_fd = ASSERT_FD(socket(AF_UNIX, SOCK_STREAM|SOCK_CLOEXEC|SOCK_NONBLOCK, 0));
+        ASSERT_OK(connect(conn2_fd, &sa.sa, salen));
+
+        ASSERT_OK(af_unix_get_qlen(unix_fd, &q));
+        ASSERT_EQ(q, 2U);
+
+        _cleanup_close_ int efd = ASSERT_FD(eventfd(0, EFD_CLOEXEC));
+        ASSERT_ERROR(af_unix_get_qlen(efd, &q), ENOTSOCK);
+}
+
 DEFINE_TEST_MAIN(LOG_DEBUG);

test/integration-tests/README.md

@@ -38,14 +38,14 @@ directory (`OutputDirectory=`) to point to the other directory using `mkosi/mkos
 After the image has been built, the integration tests can be run with:
 
 ```shell
-$ env SYSTEMD_INTEGRATION_TESTS=1 mkosi -f sandbox -- meson test -C build --suite integration-tests --num-processes "$(($(nproc) / 4))"
+$ mkosi -f sandbox -- meson test -C build --setup=integration --suite integration-tests --num-processes "$(($(nproc) / 4))"
 ```
 
 As usual, specific tests can be run in meson by appending the name of the test
 which is usually the name of the directory e.g.
 
 ```shell
-$ env SYSTEMD_INTEGRATION_TESTS=1 mkosi -f sandbox -- meson test -C build -v TEST-01-BASIC
+$ mkosi -f sandbox -- meson test -C build --setup=integration -v TEST-01-BASIC
 ```
 
 See `mkosi -f sandbox -- meson introspect build --tests` for a list of tests.
@@ -55,7 +55,7 @@ To interactively debug a failing integration test, the `--interactive` option
 newer:
 
 ```shell
-$ env SYSTEMD_INTEGRATION_TESTS=1 mkosi -f sandbox -- meson test -C build -i TEST-01-BASIC
+$ mkosi -f sandbox -- meson test -C build --setup=integration -i TEST-01-BASIC
 ```
 
 Due to limitations in meson, the integration tests do not yet depend on the
@@ -64,7 +64,7 @@ running the integration tests. To rebuild the image and rerun a test, the
 following command can be used:
 
 ```shell
-$ mkosi -f sandbox -- meson compile -C build mkosi && env SYSTEMD_INTEGRATION_TESTS=1 mkosi -f sandbox -- meson test -C build -v TEST-01-BASIC
+$ mkosi -f sandbox -- meson compile -C build mkosi && mkosi -f sandbox -- meson test -C build --setup=integration -v TEST-01-BASIC
 ```
 
 The integration tests use the same mkosi configuration that's used when you run
@@ -78,7 +78,7 @@ To iterate on an integration test, let's first get a shell in the integration te
 the following:
 
 ```shell
-$ mkosi -f sandbox -- meson compile -C build mkosi && env SYSTEMD_INTEGRATION_TESTS=1 TEST_SHELL=1 mkosi -f sandbox -- meson test -C build -i TEST-01-BASIC
+$ mkosi -f sandbox -- meson compile -C build mkosi && mkosi -f sandbox -- meson test -C build --setup=shell -i TEST-01-BASIC
 ```
 
 This will get us a shell in the integration test environment after booting the machine without running the
@@ -107,7 +107,7 @@ re-running the test will first install the new packages we just built, make a ne
 the test again. You can keep running the loop of `mkosi -R`, `systemctl soft-reboot` and
 `systemctl start ...` until the changes to the integration test are working.
 
-If you're debugging a failing integration test (running `meson test --interactive` without `TEST_SHELL`),
+If you're debugging a failing integration test (running `meson test --interactive`),
 there's no need to run `systemctl start ...`, running `systemctl soft-reboot` on its own is sufficient to
 rerun the test.
 
@@ -120,10 +120,6 @@ rerun the test.
 `TEST_NO_KVM=1`: Disable qemu KVM auto-detection (may be necessary when you're
 trying to run the *vanilla* qemu and have both qemu and qemu-kvm installed)
 
-`TEST_SHELL=1`: Configure the machine to be more *user-friendly* for
-interactive debugging (e.g. by setting a usable default terminal, suppressing
-the shutdown after the test, etc.).
-
 `TEST_MATCH_SUBTEST=subtest`:  If the test makes use of `run_subtests` use this
 variable to provide a POSIX extended regex to run only subtests matching the
 expression.

test/integration-tests/integration-test-wrapper.py

@@ -361,7 +361,7 @@ def statfs(path: Path) -> str:
 
 def main() -> None:
     parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument('--mkosi', required=True)
+    parser.add_argument('--mkosi', default=None)
     parser.add_argument('--meson-source-dir', required=True, type=Path)
     parser.add_argument('--meson-build-dir', required=True, type=Path)
     parser.add_argument('--name', required=True)
@@ -379,6 +379,12 @@ def main() -> None:
     parser.add_argument('mkosi_args', nargs='*')
     args = parser.parse_args()
 
+    if not args.mkosi:
+        args.mkosi = shutil.which('mkosi')
+        if not args.mkosi:
+            print('Could not find mkosi which is required to run the integration tests', file=sys.stderr)
+            sys.exit(1)
+
     # The meson source directory can either be the top-level repository directory or the
     # test/integration-tests/standalone subdirectory in the repository directory. The mkosi configuration
     # will always be a parent directory of one of these directories and at most 4 levels upwards, so don't
@@ -395,13 +401,6 @@ def main() -> None:
         )
         exit(1)
 
-    if not bool(int(os.getenv('SYSTEMD_INTEGRATION_TESTS', '0'))):
-        print(
-            f'SYSTEMD_INTEGRATION_TESTS=1 not found in environment, skipping {args.name}',
-            file=sys.stderr,
-        )
-        exit(77)
-
     if args.slow and not bool(int(os.getenv('SYSTEMD_SLOW_TESTS', '0'))):
         print(
             f'SYSTEMD_SLOW_TESTS=1 not found in environment, skipping {args.name}',

test/integration-tests/meson.build

@@ -1,5 +1,9 @@
 # SPDX-License-Identifier: LGPL-2.1-or-later
 
+# We'd give these more descriptive names but only alphanumeric characters are allowed.
+add_test_setup('integration')
+add_test_setup('shell', env : {'TEST_SHELL' : '1'})
+
 integration_test_wrapper = find_program('integration-test-wrapper.py')
 integration_tests = []
 integration_test_template = {
@@ -129,11 +133,11 @@ foreach integration_test : integration_tests
                 integration_test_args += ['--skip']
         endif
 
-        if not mkosi.found()
-                continue
+        if mkosi.found()
+                integration_test_args += ['--mkosi', mkosi.full_path()]
         endif
 
-        integration_test_args += ['--mkosi', mkosi.full_path(), '--']
+        integration_test_args += ['--']
 
         if integration_test['cmdline'].length() > 0
                 integration_test_args += [
@@ -151,19 +155,12 @@ foreach integration_test : integration_tests
 
         integration_test_args += integration_test['mkosi-args']
 
-        integration_test_env = {}
-
-        if want_integration_tests
-                integration_test_env += {'SYSTEMD_INTEGRATION_TESTS': '1'}
-        endif
-
         # We don't explicitly depend on the "mkosi" target because that means the image is rebuilt on every
         # "ninja -C build". Instead, the mkosi target has to be rebuilt manually before running the
         # integration tests with mkosi.
         test(
                 integration_test['name'],
                 integration_test_wrapper,
-                env : integration_test_env,
                 args : integration_test_args,
                 timeout : integration_test['timeout'],
                 priority : integration_test['priority'],

test/integration-tests/standalone/meson.build

@@ -16,7 +16,6 @@ project('systemd-testsuite',
 
 fs = import('fs')
 mkosi = find_program('mkosi', required : true)
-want_integration_tests = true
 
 # meson refuses .. in subdir() so we use a symlink to trick it into accepting it anyway.
 subdir('integration-tests')

test/units/TEST-04-JOURNAL.LogFilterPatterns.sh

@@ -36,6 +36,7 @@ run_service_and_fetch_logs() {
     fi
 
     systemctl start "$unit"
+    journalctl --sync
     journalctl -q -u "$unit" -I -p notice
 }
 

test/units/TEST-07-PID1.startv.sh

@@ -0,0 +1,39 @@
+#!/usr/bin/env bash
+# SPDX-License-Identifier: LGPL-2.1-or-later
+# shellcheck disable=SC2016
+set -eux
+set -o pipefail
+
+# shellcheck source=test/units/util.sh
+. "$(dirname "$0")"/util.sh
+
+systemd-run -v --wait echo wampfl | grep wampfl
+
+systemd-run -v -p Type=notify bash -c 'echo brumfl ; systemd-notify --ready ; echo krass' |  grep brumfl
+
+mkdir -p /run/systemd/journald.conf.d/
+
+# Let's disable storage of debug messages, since we want to flood the journal
+# daemon with messages that it will have to process, but we do not actually
+# want to push out our own messages from storage while doing so
+cat >> /run/systemd/journald.conf.d/50-disable-debug.conf <<EOF
+[Journal]
+MaxLevelStore=info
+EOF
+
+systemctl restart systemd-journald
+
+# Now flood the journal via syslog and the stream transport to ensure this finishes correctly even if busy
+( xxd /dev/urandom | logger -p debug ) &
+( xxd /dev/urandom | systemd-cat -p debug ) &
+
+# Verify that this works even if the journal is super busy
+systemd-run -v -p Type=notify bash -c 'echo schmurz ; systemd-notify --ready ; echo kropf' |  grep schmurz
+
+kill %1
+kill %2
+
+rm /run/systemd/journald.conf.d/50-disable-debug.conf
+rmdir /run/systemd/journald.conf.d ||:
+
+systemctl restart systemd-journald

units/systemd-journald-varlink@.socket

@@ -15,4 +15,4 @@ StopWhenUnneeded=yes
 [Socket]
 Service=systemd-journald@%i.service
 ListenStream=/run/systemd/journal.%i/io.systemd.journal
-SocketMode=0600
+SocketMode=0666