queue: Fix disk queue corruption detection and memory safety

.github/workflows/run_checks.yml

@@ -25,7 +25,7 @@
 name: check
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
   cancel-in-progress: true
 
 on:
@@ -41,7 +41,7 @@
     strategy:
       fail-fast: true
       matrix:
         config: [clang9, clang18-noatomics, clang18-ndebug, gcc8-debug, gcc-11-ndebug, gcc14-gnu23-debug]
     steps:
       - name: git checkout project
         uses: actions/checkout@v4
@@ -69,33 +69,33 @@
           export CFLAGS='-g'
           case "${{ matrix.config }}" in
           'clang9')
               export RSYSLOG_DEV_CONTAINER='rsyslog/rsyslog_dev_base_ubuntu:20.04'
               export CC='clang-9'
               ;;
           'clang18-ndebug')
               export RSYSLOG_CONFIGURE_OPTIONS_EXTRA='--enable-debug=no'
               export RSYSLOG_DEV_CONTAINER='rsyslog/rsyslog_dev_base_ubuntu:24.04'
               export CC='clang-18'
               ;;
           'clang18-noatomics')
               export RSYSLOG_CONFIGURE_OPTIONS_EXTRA='--enable-atomic-operations=no'
               export RSYSLOG_DEV_CONTAINER='rsyslog/rsyslog_dev_base_ubuntu:24.04'
               export CC='clang-18'
               ;;
           'gcc8-debug')
               export RSYSLOG_CONFIGURE_OPTIONS_EXTRA='--enable-debug=yes'
               export RSYSLOG_DEV_CONTAINER='rsyslog/rsyslog_dev_base_ubuntu:18.04'
               export CC='gcc-8'
               ;;
           'gcc-11-ndebug')
               export RSYSLOG_CONFIGURE_OPTIONS_EXTRA='--enable-debug=no'
               export RSYSLOG_DEV_CONTAINER='rsyslog/rsyslog_dev_base_ubuntu:22.04'
               export CC='gcc-11'
               ;;
           'gcc14-gnu23-debug')
               # omamqp1 seems to have an issue with the build system - exclude it for now
               # rgerhards, 2024-12-06
               export RSYSLOG_CONFIGURE_OPTIONS_EXTRA='--enable-debug=yes --disable-omamqp1'
               export RSYSLOG_DEV_CONTAINER='rsyslog/rsyslog_dev_base_ubuntu:24.04'
               export CFLAGS="-g -std=gnu23"
               export CC='gcc-14'
@@ -1568,6 +1568,7 @@
           chmod -R go+rw .
           docker run --rm --platform ${{ matrix.platform }} \
             -e ARCH=${{ matrix.arch }} \
+            --privileged \
             --cap-add SYS_ADMIN \
             --cap-add SYS_PTRACE \
             --security-opt seccomp=unconfined \
@@ -1578,6 +1579,12 @@
               export CFLAGS="-g -O1 -fno-omit-frame-pointer"
               export CC=gcc
               export TEST_MAX_RUNTIME=1200
+              ulimit -c unlimited
+              if [ "$ARCH" = "arm64" ]; then
+                sysctl -w kernel.core_pattern=/rsyslog/tests/core.%e.%p
+                sysctl -w kernel.core_uses_pid=1
+                echo "core_pattern=$(cat /proc/sys/kernel/core_pattern)"
+              fi
               autoreconf -fvi
               if [ "$ARCH" = "armhf" ]; then
                 ./configure --enable-silent-rules --enable-testbench \
@@ -1609,7 +1616,7 @@
               else
                 export CFLAGS="-g -O1 -fno-omit-frame-pointer -fsanitize=address -fsanitize-address-use-after-scope"
                 export LDFLAGS="-fsanitize=address"
-                export ASAN_OPTIONS="abort_on_error=1:symbolize=1:detect_leaks=0"
+                export ASAN_OPTIONS="abort_on_error=1:symbolize=1:detect_leaks=0:disable_coredump=0"
                 ./configure --enable-silent-rules --enable-testbench \
                   --enable-imdiag --disable-imdocker \
                   --enable-imfile --disable-imfile-tests \
@@ -1654,6 +1661,7 @@
           path: |
             tests/test-suite.log
             failed-tests.log
+            tests/core*
           retention-days: 7
           if-no-files-found: ignore
 
@@ -1666,3 +1674,167 @@
       - name: Skip ${{ matrix.arch }} CI, no relevant changes
         if: steps.code_changes.outputs.any_changed != 'true'
         run: echo "No relevant changes detected; skipping ${{ matrix.arch }} CI."
+
+  arm64_diskqueue_debug:
+    needs: compile
+    if: ${{ needs.compile.result == 'success' }}
+    permissions:
+      contents: read
+    runs-on: ubuntu-24.04-arm
+    timeout-minutes: 90
+    name: arm64 diskqueue debug
+    steps:
+      - name: git checkout project
+        uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      - name: fetch upstream (for changed-files diff)
+        run: |
+          git remote add upstream https://github.com/${{ github.repository }}.git
+          git fetch upstream "${{ github.event.pull_request.base.ref }}"
+
+      - name: Check for relevant changes
+        id: code_changes
+        uses: tj-actions/changed-files@v46
+        with:
+          base_sha: ${{ github.event.pull_request.base.sha }}
+          sha: ${{ github.event.pull_request.head.sha }}
+          files: |
+            **/*.c
+            **/*.h
+            tests/diskqueue-oncorruption-missing-segment.sh
+            tests/diag.sh
+            **/Makefile.am
+            configure.ac
+            .github/workflows/run_checks.yml
+            devtools/ci/Dockerfile.arm
+          files_ignore: |
+            doc/Makefile.am
+
+      - name: Set up Docker Buildx
+        if: steps.code_changes.outputs.any_changed == 'true'
+        uses: docker/setup-buildx-action@v3
+
+      - name: Build arm64 debug image (cached)
+        if: steps.code_changes.outputs.any_changed == 'true'
+        uses: docker/build-push-action@v6
+        with:
+          context: .
+          file: devtools/ci/Dockerfile.arm
+          platforms: linux/arm64
+          tags: rsyslog-arm-debug:arm64
+          load: true
+          cache-from: type=gha,scope=arm-debug-arm64
+          cache-to: type=gha,mode=max,scope=arm-debug-arm64
+
+      - name: Run arm64 diskqueue debug testcase
+        if: steps.code_changes.outputs.any_changed == 'true'
+        run: |
+          chmod -R go+rw .
+          docker run --rm --platform linux/arm64 \
+            -e HOST_UID="$(id -u)" \
+            -e HOST_GID="$(id -g)" \
+            --privileged \
+            --cap-add SYS_ADMIN \
+            --cap-add SYS_PTRACE \
+            --security-opt seccomp=unconfined \
+            -v "${{ github.workspace }}:/rsyslog" \
+            -w /rsyslog \
+            rsyslog-arm-debug:arm64 bash -c '
+              set -e
+              umask 022
+              export CFLAGS="-g -O1 -fno-omit-frame-pointer -fsanitize=address -fsanitize-address-use-after-scope"
+              export LDFLAGS="-fsanitize=address"
+              export ASAN_OPTIONS="abort_on_error=1:symbolize=1:detect_leaks=0:disable_coredump=0"
+              export TEST_MAX_RUNTIME=1200
+              export USE_AUTO_DEBUG=off
+              ulimit -c unlimited
+              sysctl -w kernel.core_pattern=/rsyslog/tests/core.%e.%p
+              sysctl -w kernel.core_uses_pid=1
+              echo "core_pattern=$(cat /proc/sys/kernel/core_pattern)"
+              autoreconf -fvi
+              ./configure --enable-silent-rules --enable-testbench \
+                --enable-imdiag --disable-imdocker \
+                --enable-imfile --disable-imfile-tests \
+                --enable-impstats --enable-imptcp \
+                --enable-mmanon --enable-mmaudit \
+                --enable-mmfields --enable-mmjsonparse \
+                --enable-mmpstrucdata --enable-mmsequence \
+                --enable-mmutf8fix --enable-mail \
+                --enable-omprog --enable-improg \
+                --enable-omruleset --enable-omstdout \
+                --enable-omuxsock \
+                --disable-pmnormalize \
+                --enable-pmaixforwardedfrom --enable-pmciscoios \
+                --enable-pmcisconames --enable-pmlastmsg \
+                --enable-pmsnare --enable-libgcrypt \
+                --disable-mmnormalize --disable-omudpspoof \
+                --enable-relp --enable-mmsnmptrapd \
+                --enable-gnutls --enable-usertools \
+                --disable-mysql \
+                --disable-valgrind --disable-mmkubernetes \
+                --disable-omkafka --disable-imkafka \
+                --disable-ommongodb --disable-omrabbitmq \
+                --disable-mmdarwin \
+                --disable-helgrind --enable-uuid \
+                --disable-fmhttp \
+                --disable-elasticsearch-tests \
+                --disable-kafka-tests --disable-snmp-tests
+              make -j8
+              set +e
+              ./tests/diskqueue-oncorruption-missing-segment.sh
+              test_rc=$?
+              set -e
+              find tests -maxdepth 1 -type f \
+                \( -name "arm64-diskqueue.debuglog" \
+                -o -name "log" \
+                -o -name "rstb_*" \
+                -o -name "core*" \
+                -o -name "failed-tests.log" \) \
+                -exec chown "$HOST_UID:$HOST_GID" {} + || true
+              find tests -maxdepth 1 -type f \
+                \( -name "arm64-diskqueue.debuglog" \
+                -o -name "log" \
+                -o -name "rstb_*" \
+                -o -name "core*" \
+                -o -name "failed-tests.log" \) \
+                -exec chmod a+r {} + || true
+              exit "$test_rc"
+            '
+
+      - name: Collect test logs
+        if: ${{ always() && steps.code_changes.outputs.any_changed == 'true' }}
+        run: devtools/gather-check-logs.sh || true
+
+      - name: Upload arm64 diskqueue debug artifacts
+        if: ${{ always() && steps.code_changes.outputs.any_changed == 'true' }}
+        uses: actions/upload-artifact@v4
+        with:
+          name: arm64-diskqueue-debug-logs
+          path: |
+            tests/failed-tests.log
+            failed-tests.log
+            tests/log
+            tests/arm64-diskqueue.debuglog
+            tests/rstb_*.log
+            tests/rstb_*.rsyslogd.log
+            tests/rstb_*.debuglog
+            tests/rstb_*.started
+            tests/core*
+          retention-days: 7
+          if-no-files-found: ignore
+
+      - name: show error logs (if we errored)
+        if: ${{ (failure() || cancelled()) && steps.code_changes.outputs.any_changed == 'true' }}
+        run: |
+          devtools/gather-check-logs.sh
+          cat failed-tests.log
+          if [ -f tests/log ]; then
+            echo "===== tail -n 2000 tests/log ====="
+            tail -n 2000 tests/log
+          fi
+
+      - name: Skip arm64 diskqueue debug, no relevant changes
+        if: steps.code_changes.outputs.any_changed != 'true'
+        run: echo "No relevant changes detected; skipping arm64 diskqueue debug."

action.c

@@ -111,6 +111,9 @@ static rsRetVal doSubmitToActionQComplex(action_t *const pAction, wti_t *const p
 static rsRetVal doSubmitToActionQNotAllMark(action_t *const pAction, wti_t *const pWti, smsg_t *);
 static void ATTR_NONNULL() actionSuspend(action_t *const pThis, wti_t *const pWti);
 static void ATTR_NONNULL() actionRetry(action_t *const pThis, wti_t *const pWti);
+static inline int wtiShutdownImmediate(const wti_t *const pWti) {
+    return ATOMIC_FETCH_32BIT(pWti->pbShutdownImmediate, pWti->pmutShutdownImmediate);
+}
 
 /* object static data (once for all instances) */
 DEFobjCurrIf(obj) DEFobjCurrIf(datetime) DEFobjCurrIf(module) DEFobjCurrIf(statsobj) DEFobjCurrIf(ruleset)
@@ -822,7 +825,7 @@ static rsRetVal ATTR_NONNULL() actionDoRetry(action_t *const pThis, wti_t *const
     assert(pThis != NULL);
 
     iRetries = 0;
-    while ((*pWti->pbShutdownImmediate == 0) && getActionState(pWti, pThis) == ACT_STATE_RTRY) {
+    while ((wtiShutdownImmediate(pWti) == 0) && getActionState(pWti, pThis) == ACT_STATE_RTRY) {
         if (actionIsDisabled(pThis)) {
             break;
         }
@@ -865,7 +868,7 @@ static rsRetVal ATTR_NONNULL() actionDoRetry(action_t *const pThis, wti_t *const
                     pThis->pszName, pThis->iResumeInterval, (long long)pThis->ttResumeRtry, (long long)ttTemp,
                     iRetries);
                 srSleep(pThis->iResumeInterval, 0);
-                if (*pWti->pbShutdownImmediate) {
+                if (wtiShutdownImmediate(pWti)) {
                     ABORT_FINALIZE(RS_RET_FORCE_TERM);
                 }
             }
@@ -895,7 +898,7 @@ static rsRetVal ATTR_NONNULL() actionDoRetry_extFile(action_t *const pThis, wti_
 
     DBGPRINTF("actionDoRetry_extFile: enter, actionState: %d\n", getActionState(pWti, pThis));
     iRetries = 0;
-    while ((*pWti->pbShutdownImmediate == 0) && getActionState(pWti, pThis) == ACT_STATE_RTRY) {
+    while ((wtiShutdownImmediate(pWti) == 0) && getActionState(pWti, pThis) == ACT_STATE_RTRY) {
         if (actionIsDisabled(pThis)) {
             break;
         }
@@ -926,7 +929,7 @@ static rsRetVal ATTR_NONNULL() actionDoRetry_extFile(action_t *const pThis, wti_
             } else {
                 ++iRetries;
                 srSleep(pThis->iResumeInterval, 0);
-                if (*pWti->pbShutdownImmediate) {
+                if (wtiShutdownImmediate(pWti)) {
                     ABORT_FINALIZE(RS_RET_FORCE_TERM);
                 }
             }
@@ -1784,7 +1787,7 @@ static rsRetVal ATTR_NONNULL() processBatchMain(void *__restrict__ const pVoid,
         ABORT_FINALIZE(RS_RET_DISABLE_ACTION);
     }
 
-    for (i = 0; i < batchNumMsgs(pBatch) && !*pWti->pbShutdownImmediate; ++i) {
+    for (i = 0; i < batchNumMsgs(pBatch) && !wtiShutdownImmediate(pWti); ++i) {
         if (batchIsValidElem(pBatch, i)) {
             /* we do not check error state below, because aborting would be
              * more harmful than continuing.

devtools/ci/Dockerfile.arm

@@ -7,5 +7,5 @@ RUN apt-get update \
        libgnutls28-dev libestr-dev libfastjson-dev zlib1g-dev \
        libgcrypt20-dev librelp-dev uuid-dev libyaml-dev \
        libcurl4-gnutls-dev libprotobuf-c-dev protobuf-c-compiler libsnappy-dev \
-       iproute2 \
+       iproute2 gdb procps \
     && rm -rf /var/lib/apt/lists/*

doc/source/rainerscript/queue_parameters.rst

@@ -318,6 +318,38 @@ be turned on without a good reason. Note that the penalty also depends on
 *queue.checkpointInterval* frequency.
 
 
+queue.onCorruption
+------------------
+
+.. csv-table::
+   :header: "type", "default", "mandatory", "|FmtObsoleteName| directive"
+   :widths: auto
+   :class: parameter-table
+
+   "word", "safe", "no", "none"
+
+Controls how rsyslog handles disk queue corruption detected during startup.
+This applies to queue files and checkpoint state (``.qi``).
+
+Supported values are:
+
+* ``safe``: detect corruption, move queue files to a timestamped
+  ``.bad`` directory, and start with a fresh disk queue.
+* ``inMemory``: switch to a non-persistent in-memory queue for the current
+  process lifetime.
+* ``ignore``: skip the corruption verification logic and keep legacy startup
+  behavior.
+
+If rsyslog cannot safely quarantine corrupted files in ``safe`` mode (for
+example because the recovery directory cannot be created or files cannot be
+moved), it logs alert-level errors and switches to pure in-memory emergency
+mode for safety.
+
+The startup verification checks queue structure consistency (file sequence and
+pointer continuity). It does **not** parse and validate each payload record
+inside queue segment files.
+
+
 queue.samplingInterval
 ----------------------