in_storage_backlog: Fix persistent chunks issues without matching routes on hot reloading

AGENTS.md

@@ -101,30 +101,37 @@ Keep changes scoped: plugin logic in its plugin directory, shared behavior in `s
 - For touched components covered by `tests/integration`, agents must run the
   focused scenario(s) twice:
   - once normally to verify behavior;
-  - once with valgrind enabled to verify memory-safety behavior.
+  - once with the platform memory checker enabled to verify memory-safety
+    behavior: Valgrind on Linux or Leaks on macOS.
 - The default expectation for component verification is:
   `./tests/integration/setup-venv.sh`
   `cmake -S . -B build -DFLB_TESTS_RUNTIME=On -DFLB_TESTS_INTERNAL=On`
   `cmake --build build -j8`
   `tests/integration/.venv/bin/python -m pytest <focused-scenario> -q`
+  On Linux, run the memory-safety pass with:
   `VALGRIND=1 VALGRIND_STRICT=1 tests/integration/.venv/bin/python -m pytest <focused-scenario> -q`
+  On macOS, run the memory-safety pass with:
+  `LEAKS=1 LEAKS_STRICT=1 tests/integration/.venv/bin/python -m pytest <focused-scenario> -q`
 - On Windows, use the same `-DFLB_TESTS_RUNTIME=On` configuration and run
-  relevant focused runtime cases. Valgrind is normally unavailable on Windows;
-  report that exact blocker instead of conflating it with runtime-test support.
+  relevant focused runtime and functional integration cases. Valgrind and
+  macOS Leaks are normally unavailable on Windows; report that exact
+  memory-checker blocker instead of conflating it with test support.
 - Run broader test coverage when changing shared lifecycle, routing, storage, or accounting code.
 - Validate both success and failure paths (invalid payloads, boundary sizes, null/missing fields).
 - You can also run specific binaries from `build/bin` (e.g., `./bin/flb-it-opentelemetry`).
 - When changing code covered by `tests/integration`, agents must verify the
-  affected scenarios are valgrind-clean. Run the focused integration tests with
-  `tests/integration/run_tests.py --valgrind --valgrind-strict ...` and do not
-  stop at functional pass/fail if memory errors or leaks remain.
-- If a focused integration or valgrind run cannot be executed, agents must not
-  silently skip it. They must report the exact blocker in the final response
-  (for example: missing binary, missing Python environment, unsupported
-  scenario, missing dependency, or infrastructure failure).
+  affected scenarios are clean under the platform memory checker. On Linux,
+  run `tests/integration/run_tests.py --valgrind --valgrind-strict ...`. On
+  macOS, run `tests/integration/run_tests.py --leaks --leaks-strict ...`. Do
+  not stop at functional pass/fail if memory errors or leaks remain.
+- If a focused integration or platform memory-checker run cannot be executed,
+  agents must not silently skip it. They must report the exact blocker in the
+  final response (for example: missing binary, missing Python environment,
+  unsupported scenario, missing dependency, or infrastructure failure).
 - Final task close-outs must include proof of verification:
   - the exact focused integration command(s) run;
-  - whether valgrind was used;
+  - which platform memory checker was used (Valgrind on Linux or Leaks on
+    macOS), or `not run` with the exact blocker;
   - pass/fail status;
   - any concrete blocker if a required run could not be completed.
 - Keep generated integration artifacts out of git. Do not commit

plugins/in_storage_backlog/sb.c

@@ -312,8 +312,11 @@ static int sb_append_chunk_to_segregated_backlogs(struct cio_chunk  *target_chun
         return -2;
     }
 
-    flb_routes_mask_set_by_tag(dummy_input_chunk.routes_mask, tag_buf, tag_len,
-                               context->ins);
+    result = flb_routes_mask_set_by_tag(dummy_input_chunk.routes_mask, tag_buf, tag_len,
+                                        context->ins);
+    if (result == 0) {
+        return -4;
+    }
 
     mk_list_foreach_safe(head, tmp, &context->backlogs) {
         backlog = mk_list_entry(head, struct sb_out_queue, _head);
@@ -412,6 +415,20 @@ int sb_segregate_chunks(struct flb_config *config)
             /* try to segregate a chunk */
             ret = sb_append_chunk_to_segregated_backlogs(chunk, stream, context);
             if (ret) {
+                /*
+                 * Leave chunks without a route on disk so a future configuration
+                 * can reconsider them. Closing the ChunkIO handle removes them
+                 * from this context's active chunk accounting without deleting
+                 * the underlying files.
+                 */
+                if (ret == -4) {
+                    flb_plg_info(context->ins,
+                                 "no matching route for %s/%s, keeping it on disk",
+                                 stream->name, chunk->name);
+                    cio_chunk_close(chunk, CIO_FALSE);
+                    continue;
+                }
+
                 /*
                  * if the chunk could not be segregated, just remove it from the
                  * queue, delete it and continue.

tests/integration/scenarios/in_storage_backlog/config/reload_with_stale_route.yaml

@@ -0,0 +1,19 @@
+service:
+  flush: 1
+  log_level: info
+  http_server: on
+  http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT}
+  hot_reload: on
+  hot_reload.timeout: 10
+  storage.path: ${STORAGE_BACKLOG_PATH}
+
+pipeline:
+  inputs:
+    - name: tail
+      tag: live
+      path: ${STORAGE_BACKLOG_MISSING_PATH}
+      storage.type: filesystem
+
+  outputs:
+    - name: null
+      match: "*"

tests/integration/scenarios/in_storage_backlog/config/reload_without_stale_route.yaml

@@ -0,0 +1,19 @@
+service:
+  flush: 1
+  log_level: info
+  http_server: on
+  http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT}
+  hot_reload: on
+  hot_reload.timeout: 10
+  storage.path: ${STORAGE_BACKLOG_PATH}
+
+pipeline:
+  inputs:
+    - name: tail
+      tag: live
+      path: ${STORAGE_BACKLOG_MISSING_PATH}
+      storage.type: filesystem
+
+  outputs:
+    - name: null
+      match: live

tests/integration/scenarios/in_storage_backlog/config/seed_unroutable_chunk.yaml

@@ -0,0 +1,18 @@
+service:
+  flush: 1
+  grace: 0
+  storage.path: ${STORAGE_BACKLOG_PATH}
+
+pipeline:
+  inputs:
+    - name: dummy
+      tag: stale
+      samples: 1
+      storage.type: filesystem
+
+  outputs:
+    - name: http
+      match: stale
+      host: 127.0.0.1
+      port: 9
+      retry_limit: no_limits

tests/integration/scenarios/in_storage_backlog/tests/test_in_storage_backlog_001.py

@@ -0,0 +1,174 @@
+import os
+from pathlib import Path
+import signal
+import shutil
+import subprocess
+import time
+
+import pytest
+
+from utils.fluent_bit_manager import FluentBitManager
+
+
+CONFIG_DIR = Path(__file__).resolve().parent.parent / "config"
+SEED_CONFIG = CONFIG_DIR / "seed_unroutable_chunk.yaml"
+RELOAD_WITHOUT_ROUTE = CONFIG_DIR / "reload_without_stale_route.yaml"
+RELOAD_WITH_ROUTE = CONFIG_DIR / "reload_with_stale_route.yaml"
+KEEP_ON_DISK_MESSAGE = "no matching route for dummy.0/"
+REGISTER_MESSAGE = "register dummy.0/"
+QUEUE_MESSAGE = "queueing dummy.0:"
+
+
+def wait_for_persisted_chunk(storage_path, process, log_path, timeout=10):
+    deadline = time.time() + timeout
+
+    while time.time() < deadline:
+        chunks = list(storage_path.glob("*/*.flb"))
+        if chunks and chunks[0].stat().st_size > 0:
+            return chunks[0]
+
+        return_code = process.poll()
+        if return_code is not None:
+            if log_path.exists():
+                log = log_path.read_text(encoding="utf-8", errors="replace")
+            else:
+                log = "seed log was not created"
+            raise RuntimeError(
+                f"seed Fluent Bit exited with code {return_code} before persisting a chunk:\n{log}"
+            )
+
+        time.sleep(0.1)
+
+    raise TimeoutError("timed out waiting for the seed filesystem chunk")
+
+
+def wait_for_log_occurrences(log_path, text, expected, timeout=10):
+    deadline = time.time() + timeout
+
+    while time.time() < deadline:
+        if log_path.exists():
+            log = log_path.read_text(encoding="utf-8", errors="replace")
+            if log.count(text) >= expected:
+                return
+
+        time.sleep(0.1)
+
+    raise TimeoutError(f"timed out waiting for {expected} occurrences of {text!r}")
+
+
+def wait_for_path_removal(path, timeout=10):
+    deadline = time.time() + timeout
+
+    while time.time() < deadline:
+        if not path.exists():
+            return
+
+        time.sleep(0.1)
+
+    raise TimeoutError(f"timed out waiting for {path} to be removed")
+
+
+def seed_persisted_chunk(storage_path, seed_log_path):
+    binary_path = FluentBitManager(str(SEED_CONFIG)).binary_absolute_path
+    seed_process = subprocess.Popen(
+        [binary_path, "-c", str(SEED_CONFIG), "-l", str(seed_log_path)],
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.STDOUT,
+        env=os.environ.copy(),
+    )
+
+    try:
+        return wait_for_persisted_chunk(storage_path, seed_process, seed_log_path)
+    finally:
+        if seed_process.poll() is None:
+            seed_process.kill()
+        seed_process.wait(timeout=10)
+
+
+@pytest.fixture
+def seeded_chunk(tmp_path, monkeypatch):
+    storage_path = tmp_path / "storage"
+    storage_path.mkdir()
+    seed_log_path = tmp_path / "seed.log"
+    missing_path = tmp_path / "does-not-exist.log"
+
+    monkeypatch.setenv("STORAGE_BACKLOG_PATH", str(storage_path))
+    monkeypatch.setenv("STORAGE_BACKLOG_MISSING_PATH", str(missing_path))
+
+    persisted_chunk = seed_persisted_chunk(storage_path, seed_log_path)
+    assert persisted_chunk.exists()
+
+    return persisted_chunk
+
+
+def test_routable_persisted_chunk_is_replayed_on_startup(seeded_chunk):
+    manager = FluentBitManager(str(RELOAD_WITH_ROUTE))
+
+    try:
+        manager.start()
+        wait_for_log_occurrences(
+            Path(manager.log_file), REGISTER_MESSAGE, expected=1
+        )
+        wait_for_log_occurrences(
+            Path(manager.log_file), QUEUE_MESSAGE, expected=1
+        )
+        wait_for_path_removal(seeded_chunk)
+    finally:
+        manager.stop()
+
+    assert not seeded_chunk.exists()
+
+
+@pytest.mark.skipif(not hasattr(signal, "SIGHUP"), reason="SIGHUP is unavailable")
+def test_unroutable_chunk_stays_on_disk_across_hot_reload(seeded_chunk):
+    manager = FluentBitManager(str(RELOAD_WITHOUT_ROUTE))
+
+    try:
+        manager.start()
+        wait_for_log_occurrences(
+            Path(manager.log_file), KEEP_ON_DISK_MESSAGE, expected=1
+        )
+        assert seeded_chunk.exists()
+
+        manager.send_sighup()
+        manager.wait_for_hot_reload_count(1, timeout=20)
+        wait_for_log_occurrences(
+            Path(manager.log_file), KEEP_ON_DISK_MESSAGE, expected=2, timeout=20
+        )
+        assert seeded_chunk.exists()
+    finally:
+        manager.stop()
+
+    assert seeded_chunk.exists()
+
+
+@pytest.mark.skipif(not hasattr(signal, "SIGHUP"), reason="SIGHUP is unavailable")
+def test_restored_route_recovers_chunk_during_hot_reload(tmp_path, seeded_chunk):
+    runtime_config = tmp_path / "fluent-bit.yaml"
+    shutil.copyfile(RELOAD_WITHOUT_ROUTE, runtime_config)
+    manager = FluentBitManager(str(runtime_config))
+
+    try:
+        manager.start()
+        wait_for_log_occurrences(
+            Path(manager.log_file), KEEP_ON_DISK_MESSAGE, expected=1
+        )
+        assert seeded_chunk.exists()
+
+        pending_config = tmp_path / "fluent-bit.yaml.tmp"
+        shutil.copyfile(RELOAD_WITH_ROUTE, pending_config)
+        os.replace(pending_config, runtime_config)
+
+        manager.send_sighup()
+        manager.wait_for_hot_reload_count(1, timeout=20)
+        wait_for_log_occurrences(
+            Path(manager.log_file), REGISTER_MESSAGE, expected=1, timeout=20
+        )
+        wait_for_log_occurrences(
+            Path(manager.log_file), QUEUE_MESSAGE, expected=1, timeout=20
+        )
+        wait_for_path_removal(seeded_chunk, timeout=20)
+    finally:
+        manager.stop()
+
+    assert not seeded_chunk.exists()

tests/runtime/CMakeLists.txt

@@ -113,6 +113,10 @@ if (FLB_CUSTOM_CALYPTIA)
     endforeach()
 endif()
 
+if(FLB_IN_LIB AND FLB_OUT_HTTP AND FLB_OUT_NULL AND FLB_IN_STORAGE_BACKLOG)
+  FLB_RT_TEST(FLB_IN_STORAGE_BACKLOG "in_storage_backlog.c")
+endif()
+
 if(FLB_IN_EBPF)
     # Define common variables
     set(EBPF_TEST_INCLUDE_DIRS