Fix static journal facet filters (#22456)
Co-authored-by: vkalintiris <vasilis@netdata.cloud>
Costa Tsaousis committed
May 22, 2026 at 11:45 UTC
6a515000ac89e9e0f34390dd0e0bc9e76b5c483c
15 files changed
+1503
-31
.agents/sow/done/SOW-0015-20260508-static-journal-facets.md
new
+809
@@ -0,0 +1,809 @@
1
+# SOW-0015 - Static Journal Facet Filtering
2
+
3
+## Status
4
+
5
+Status: completed
6
+
7
+Sub-state: PR review rerun cleanup completed.
8
+
9
+## Requirements
10
+
11
+### Purpose
12
+
13
+Restore correct systemd journal facet filtering for static Netdata Agent builds that use the Rust journal provider, especially on distributions whose journal files contain LZ4-compressed data objects. Keep the fix small, source-verified, and isolated to the clean worktree created from `master`.
14
+
15
+### User Request
16
+
17
+The user reported that the prior session is from another server and must not be trusted as proof. The user clarified that dnf packages work because they use `libsystemd`, static installs fail, and all edits must happen in a new worktree from `master`, not in the dirty main checkout where another worker is active.
18
+
19
+### Assistant Understanding
20
+
21
+Facts:
22
+
23
+- Clean worktree: `~/src/PRs/netdata-static-journal-facets`, branch `fix/static-journal-facets`, created from refreshed `origin/master`.
24
+- The main checkout is dirty with unrelated netflow work and must not be edited.
25
+- Static builds can use the Rust journal provider: `CMakeLists.txt:201`, `CMakeLists.txt:220`, `CMakeLists.txt:2908`.
26
+- dnf builds on the user's tested server use native `libsystemd`; those are outside the failing path.
27
+
28
+Inferences:
29
+
30
+- The failing path is the Rust provider under `src/crates/jf/`, not native `sd-journal`.
31
+- The bug affects filtered/faceted queries more strongly than unfiltered scans because facet slicing builds journal matches from unique field values before scanning rows.
32
+
33
+Unknowns:
34
+
35
+- None for the accepted scope. A local RHEL 8.10 static install was available for regression validation after the first PR iteration.
36
+
37
+### Acceptance Criteria
38
+
39
+- Rust journal provider can return decompressed payloads for LZ4/XZ/Zstd compressed data objects.
40
+- Unique-value enumeration returns logical `field=value` bytes, not compressed payload bytes.
41
+- Data-object lookup used by match filters can find compressed data objects when the caller provides logical `field=value`.
42
+- Facet filter setup falls back to a full query if unique-value enumeration returns an error while constructing backend matches.
43
+- Existing uncompressed data behavior remains unchanged.
44
+- Focused Rust tests pass for `src/crates/jf/`.
45
+
46
+## Analysis
47
+
48
+Sources checked:
49
+
50
+- `.agents/skills/project-writing-collectors/SKILL.md`
51
+- `src/collectors/systemd-journal.plugin/systemd-journal.c`
52
+- `src/collectors/systemd-journal.plugin/provider/netdata_provider.h`
53
+- `src/collectors/systemd-journal.plugin/provider/rust_provider.h`
54
+- `src/crates/jf/journal_reader_ffi/src/lib.rs`
55
+- `src/crates/jf/journal_file/src/object.rs`
56
+- `src/crates/jf/journal_file/src/file.rs`
57
+- `src/crates/journal-core/src/file/object.rs`
58
+- `src/crates/jf/Cargo.toml`
59
+- `src/crates/jf/journal_file/Cargo.toml`
60
+- Official systemd journal file format documentation: `https://systemd.io/JOURNAL_FILE_FORMAT/`
61
+
62
+Current state:
63
+
64
+- `systemd-journal.c:583-607` builds native journal matches from `NSD_JOURNAL_FOREACH_UNIQUE()` values, parses them as `field=value`, then passes the same bytes to `nsd_journal_add_match()`.
65
+- `systemd-journal.c:661` skips a journal file when filters exist but no backend matches can be installed.
66
+- `journal_reader_ffi/src/lib.rs:285-305` decompresses compressed entry data for normal row scans.
67
+- `journal_reader_ffi/src/lib.rs:383-389` returns unique-value payload bytes without checking compression.
68
+- `journal_file/src/object.rs:873-890` marks LZ4/XZ/Zstd as compression methods, but only implements Zstd decompression.
69
+- `journal-core/src/file/object.rs:978-1021` already implements Zstd, LZ4, and XZ decompression in the newer shared journal core.
70
+- `journal_file/src/file.rs:548-550` uses payload matching for data-object lookup; `file.rs:191` compares raw object payload bytes to the caller-provided payload, which cannot find compressed objects when the caller provides logical `field=value`.
71
+- The official journal format says DATA objects contain `field=value` payloads, that XZ/LZ4/Zstd compression is signaled by object/header flags, and that the DATA object hash is computed from the payload. That matches the need to expose/deal with the logical payload even when the on-disk bytes are compressed.
72
+
73
+Risks:
74
+
75
+- A partial fix that only decompresses unique enumeration may still fail when the filter builder later looks up the compressed data object.
76
+- Adding decompression to hash-table lookup can affect filtered query performance on compressed buckets. The lookup is only on candidate objects in one hash bucket and is necessary for correctness.
77
+- Adding new Rust dependencies changes `src/crates/jf/Cargo.lock`; this must be validated with focused cargo tests.
78
+
79
+## Pre-Implementation Gate
80
+
81
+Status: ready
82
+
83
+Problem / root-cause model:
84
+
85
+- Static builds use the Rust journal provider path when `ENABLE_NETDATA_JOURNAL_FILE_READER` is enabled. That provider must emulate `libsystemd` APIs expected by `systemd-journal.plugin`.
86
+- Facet counters can be produced by row scans because `rsd_journal_enumerate_available_data()` already decompresses entry data before returning it.
87
+- Facet filtering uses unique-value enumeration to install backend matches. The Rust unique path currently returns raw payloads and does not decompress. On compressed systemd journal data, `parse_journal_field()` cannot reliably see `field=value`, so selected facet matches are not installed.
88
+- Even after returning decompressed unique data, filter construction must resolve the logical `field=value` back to the data object offset. The current data-object matcher compares raw on-disk payload bytes, so compressed objects remain unfindable by logical payload.
89
+- The `src/crates/jf/` decompressor only supports Zstd despite carrying LZ4/XZ flags. The newer `src/crates/journal-core/` code provides a local pattern for LZ4 and XZ support.
90
+
91
+Evidence reviewed:
92
+
93
+- `src/collectors/systemd-journal.plugin/systemd-journal.c:583-607`
94
+- `src/collectors/systemd-journal.plugin/systemd-journal.c:661`
95
+- `src/crates/jf/journal_reader_ffi/src/lib.rs:285-305`
96
+- `src/crates/jf/journal_reader_ffi/src/lib.rs:383-389`
97
+- `src/crates/jf/journal_file/src/object.rs:857-890`
98
+- `src/crates/jf/journal_file/src/file.rs:157-191`
99
+- `src/crates/journal-core/src/file/object.rs:978-1021`
100
+- `src/crates/jf/Cargo.toml:16-28`
101
+- `src/crates/jf/journal_file/Cargo.toml:7-16`
102
+- `https://systemd.io/JOURNAL_FILE_FORMAT/` sections "Structure", "Extensibility", "Data Objects", and "Reading".
103
+
104
+Affected contracts and surfaces:
105
+
106
+- Static-build systemd journal log queries and facet filters.
107
+- Rust FFI compatibility with the C plugin's `sd-journal`-like expectations.
108
+- Rust crate dependency lockfile for `src/crates/jf/`.
109
+- No public configuration, schema, or user-facing documentation surface is expected to change.
110
+
111
+Existing patterns to reuse:
112
+
113
+- Existing entry-data decompression branch in `rsd_journal_enumerate_available_data()`.
114
+- Existing LZ4/XZ/Zstd decompression implementation in `src/crates/journal-core/src/file/object.rs`.
115
+- Existing `JournalError::DecompressorError` / `UnknownCompressionMethod` handling.
116
+- Existing `PayloadMatcher` bucket visitor pattern.
117
+
118
+Risk and blast radius:
119
+
120
+- Scope is confined to the Rust journal file reader used by the static provider.
121
+- Native `libsystemd` builds should be unaffected.
122
+- Main behavioral risk is filtered lookup performance on compressed hash buckets; lookup remains bounded to one bucket.
123
+- Security risk is low; decompression must preserve existing error handling and must not panic on malformed compressed payloads.
124
+
125
+Sensitive data handling plan:
126
+
127
+- No raw logs, hostnames, IP addresses, customer identifiers, secrets, or journal payload samples will be written to durable artifacts.
128
+- SOW evidence records only generic OS/compression behavior and source file references.
129
+
130
+Implementation plan:
131
+
132
+1. Add LZ4 and XZ dependencies to the `src/crates/jf/` workspace and port the established decompression logic from `journal-core`.
133
+2. Make `rsd_journal_enumerate_available_unique()` mirror entry-data enumeration by returning decompressed payloads for compressed data objects.
134
+3. Make data-object payload matching compare decompressed payloads for compressed data objects so filter construction can find the matching object offset.
135
+4. Add focused Rust tests for LZ4 decompression and compressed payload matching.
136
+5. Run focused formatting and tests for the `src/crates/jf/` workspace.
137
+
138
+Validation plan:
139
+
140
+- `cargo fmt` in `src/crates/jf`.
141
+- `cargo test` in `src/crates/jf`.
142
+- Same-failure scan for remaining raw `payload_bytes()` returns in FFI paths.
143
+- Source review of all compressed data object reads in `src/crates/jf/`.
144
+
145
+Artifact impact plan:
146
+
147
+- AGENTS.md: no update expected; workflow rules unchanged.
148
+- Runtime project skills: no update expected; collector-writing guidance remains valid.
149
+- Specs: no update expected; this is a bug fix to match existing static-provider intent.
150
+- End-user/operator docs: no update expected; no user-facing command/config changes.
151
+- End-user/operator skills: no update expected; public AI skills are unaffected.
152
+- SOW lifecycle: this SOW tracks the work and is completed/moved with the implementation in the same commit.
153
+
154
+Open-source reference evidence:
155
+
156
+- No external mirrored repository evidence used yet. The fix is based on two in-repository implementations of the same journal format.
157
+
158
+Open decisions:
159
+
160
+- None. The user has already specified the worktree constraint and the failing implementation path.
161
+
162
+## Implications And Decisions
163
+
164
+- No user decision is currently required. The evidence points to a bounded bug fix in the static Rust provider.
165
+
166
+## Plan
167
+
168
+1. Patch `src/crates/jf/journal_file` decompression support and data payload matching.
169
+2. Patch `src/crates/jf/journal_reader_ffi` unique enumeration.
170
+3. Update `src/crates/jf` Cargo manifests/lockfile.
171
+4. Add focused tests.
172
+5. Run focused validation and update this SOW.
173
+
174
+## Execution Log
175
+
176
+### 2026-05-08
177
+
178
+- Created clean worktree from refreshed `origin/master`.
179
+- Loaded project collector-writing skill.
180
+- Verified source evidence and wrote pre-implementation gate.
181
+- Added LZ4/XZ decompression support to the legacy `src/crates/jf/journal_file` reader, reusing the newer `journal-core` implementation pattern.
182
+- Changed Rust FFI unique-value enumeration to return decompressed payloads for compressed DATA objects.
183
+- Changed DATA hash-bucket payload matching to compare decompressed payloads when the on-disk object is compressed.
184
+- Added focused LZ4 compressed payload matcher tests.
185
+- Corrected an existing filter test expectation: the test writes 5,000 iterations and 2 matching rows per iteration, so the expected filtered count is `2 * iterations`, not `2`.
186
+
187
+## Validation
188
+
189
+Acceptance criteria evidence:
190
+
191
+- LZ4/XZ/Zstd support: `src/crates/jf/journal_file/src/object.rs` now handles Zstd, LZ4 with the systemd 8-byte uncompressed-size prefix, and XZ.
192
+- Unique enumeration: `src/crates/jf/journal_reader_ffi/src/lib.rs` now mirrors entry-data enumeration and decompresses before returning payload bytes.
193
+- Match lookup: `src/crates/jf/journal_file/src/file.rs` now uses `DataPayloadMatcher`, which compares raw payload first and then decompressed payload for compressed DATA objects.
194
+- Uncompressed behavior: raw `object.get_payload() == self.payload` matching remains the first path.
195
+- Focused tests: `src/crates/jf/journal_file/src/file.rs` adds positive and negative LZ4 compressed payload matcher tests.
196
+
197
+Tests or equivalent validation:
198
+
199
+- `cargo fmt` in `src/crates/jf`: passed.
200
+- `cargo test -q` in `src/crates/jf`: passed; 4 tests passed.
201
+- `git diff --check`: passed.
202
+- `.agents/sow/audit.sh`: status/directory checks passed for this SOW; the audit reported one pre-existing sensitive-data pattern in `.agents/skills/mirror-netdata-repos/SKILL.md:112`, which is public SSH clone syntax (`git@github.com:netdata/...`) in an unrelated existing file, not sensitive data from this work.
203
+
204
+Real-use evidence:
205
+
206
+- Initial implementation was not run against the local RHEL 8.10 static install before opening the PR.
207
+- Regression validation below records live Function evidence from that static install after the reopened fix.
208
+
209
+Reviewer findings:
210
+
211
+- Initial implementation had no external reviewer pass before opening the PR; the user asked to distrust the prior session and verify locally from code.
212
+- PR review iterations found Copilot comments on `netdata/netdata#22456`; each was verified before code changes, addressed in the same SOW, replied to in-thread, and resolved after commit/push.
213
+
214
+Same-failure scan:
215
+
216
+- `rg` over `src/crates/jf` for `payload_bytes()`, `decompress()`, `enumerate_available_unique`, `enumerate_available_data`, and `find_data_offset()` found the fixed FFI paths and the fixed match lookup. Remaining raw `payload_bytes()` use in `src/crates/jf/journal_file/src/filter.rs:246` is a debug dump path, not match construction or row enumeration.
217
+
218
+Sensitive data gate:
219
+
220
+- Durable artifacts contain no raw logs, secrets, credentials, bearer tokens, SNMP communities, customer names, personal data, non-private customer-identifying IPs, private endpoints, or proprietary incident details.
221
+
222
+Artifact maintenance gate:
223
+
224
+- AGENTS.md: no update needed; workflow and project guardrails did not change.
225
+- Runtime project skills: no update needed; this did not change how agents should work on collectors.
226
+- Specs: no update needed; this bug fix restores the intended static provider behavior and does not create a new product contract.
227
+- End-user/operator docs: no update needed; no user-facing configuration, command, or workflow changed.
228
+- End-user/operator skills: no update needed; public/operator AI skills are unaffected.
229
+- SOW lifecycle: SOW status is `completed` and the file is moved to `.agents/sow/done/` with the implementation in the same commit.
230
+
231
+Specs update:
232
+
233
+- No spec update needed; behavior remains "static Rust provider should act like the native journal provider for DATA object payloads."
234
+
235
+Project skills update:
236
+
237
+- No project skill update needed; no new reusable workflow was discovered.
238
+
239
+End-user/operator docs update:
240
+
241
+- No docs update needed; this is a transparent bug fix.
242
+
243
+End-user/operator skills update:
244
+
245
+- No end-user/operator skill update needed; no public skill behavior changed.
246
+
247
+Lessons:
248
+
249
+- When fixing static-provider journal filtering, verify both the enumeration side and the data-offset lookup side. Returning decompressed `field=value` bytes is not enough if the filter builder still searches the DATA hash table by raw compressed bytes.
250
+
251
+Follow-up mapping:
252
+
253
+- No follow-up was needed for the initial fix; the later live regression is tracked in the appended regression section.
254
+
255
+## PR Review Iteration - 2026-05-08
256
+
257
+Findings:
258
+
259
+- `PRRT_kwDOAKPxd86Asq8i`, `src/crates/jf/journal_reader_ffi/src/lib.rs:396`: valid. The decompression error path returned generic `-1` and printed to stderr instead of returning `JournalError::to_error_code()`. Same-class sweep found the same pattern in `rsd_journal_enumerate_available_data()`.
260
+- `PRRT_kwDOAKPxd86Asq9F`, `src/crates/jf/journal_file/src/object.rs:899`: valid. The LZ4 branch trusted the on-disk uncompressed size prefix before `Vec::resize()`. Same-class sweep found the same LZ4 reader pattern in `src/crates/journal-core/src/file/object.rs`.
261
+- CI signal before this iteration: one Docker armv7 job failed in its `Build Image` step while the workflow run was still in progress. GitHub did not expose logs yet because the overall run had not completed; all other available failure sources were either pending or passing.
262
+
263
+Actions:
264
+
265
+- Changed compressed DATA enumeration FFI paths to return `e.to_error_code()` for decompression errors.
266
+- Added checked `u64` to `usize` conversion for the LZ4 uncompressed-size prefix.
267
+- Added a DATA payload upper bound matching systemd's `DATA_SIZE_MAX` journal importer limit: 768 MiB.
268
+- Used `try_reserve_exact()` before `Vec::resize()` so allocation failure returns `JournalError::DecompressorError` instead of panicking.
269
+- Applied the same LZ4 bounds hardening to both `src/crates/jf/journal_file` and `src/crates/journal-core`.
270
+- Added oversized LZ4 prefix regression tests in both readers.
271
+
272
+Validation:
273
+
274
+- `cargo fmt` in `src/crates/jf`: passed.
275
+- `cargo fmt` in `src/crates`: passed; unrelated formatting churn in `src/crates/netdata-plugin/rt/src/lib.rs` was removed before staging.
276
+- `cargo test -q` in `src/crates/jf`: passed; 5 tests passed.
277
+- `cargo test -q -p journal-core` in `src/crates`: passed; 19 tests passed plus the existing ignored tests.
278
+
279
+## PR Review Iteration 2 - 2026-05-08
280
+
281
+Findings:
282
+
283
+- `PRRT_kwDOAKPxd86AtF96`, `src/crates/journal-core/src/file/object.rs:1019`: valid. `try_reserve_exact()` was given a delta from capacity, but the API expects additional capacity from current length.
284
+- `PRRT_kwDOAKPxd86AtF-U`, `src/crates/jf/journal_file/src/object.rs:911`: valid. Same fallible-reserve issue in the legacy reader.
285
+- `PRRT_kwDOAKPxd86AtF-e`, `src/crates/jf/journal_file/src/object.rs:890`: valid. The Zstd streaming path still used unbounded `read_to_end()`; same-class sweep also covered XZ and `journal-core`.
286
+- `PRRT_kwDOAKPxd86AtF-n`, `src/crates/jf/journal_file/src/object.rs:927`: valid. The newly added XZ decompression branch needed a focused positive test; same-class sweep added the same coverage to `journal-core`.
287
+
288
+Actions:
289
+
290
+- Corrected fallible LZ4 reserve logic to reserve the full required final length after `clear()`.
291
+- Added bounded `read_limited_to_end()` helpers for streaming Zstd/XZ decompression.
292
+- Applied the same bounded streaming decompression to both `src/crates/jf/journal_file` and `src/crates/journal-core`.
293
+- Added fixed XZ compressed-payload fixtures and positive decompression tests for both readers without enabling the encoder feature in production dependencies.
294
+
295
+Validation:
296
+
297
+- `cargo fmt -p journal_file -p journal_reader_ffi` in `src/crates/jf`: passed.
298
+- `cargo fmt -p journal-core` in `src/crates`: passed.
299
+- `cargo test -q` in `src/crates/jf`: passed; 6 tests passed.
300
+- `cargo test -q -p journal-core` in `src/crates`: passed; 20 tests passed plus the existing ignored tests.
301
+
302
+## PR Review Iteration 3 - 2026-05-08
303
+
304
+Findings:
305
+
306
+- `PRRT_kwDOAKPxd86AtR3f`, `src/crates/jf/journal_file/src/file.rs:185`: partially valid. The slice comparison compiled in focused tests, but comparing by reference is clearer and avoids relying on implicit unsized slice comparison behavior.
307
+- `PRRT_kwDOAKPxd86AtR33`, `src/crates/jf/journal_file/src/object.rs:787`: considered and not implemented as shared code. The duplicate helper exists in a legacy separate `src/crates/jf` workspace and the newer `src/crates/journal-core` workspace, with different error types. Centralizing via `journal-common` would add unrelated dependencies to the legacy static reader for a small private helper. Keeping the helpers local is lower risk for this bug-fix PR.
308
+- `PRRT_kwDOAKPxd86AtR4H`, `src/crates/jf/journal_file/src/object.rs:796`: valid. `read_to_end()` can leave partial bytes in the reusable buffer before returning an error.
309
+- `PRRT_kwDOAKPxd86AtR4a`, `src/crates/journal-core/src/file/object.rs:901`: valid. The stream-size cap needed a focused small-cap test that does not allocate hundreds of MiB.
310
+
311
+Actions:
312
+
313
+- Changed compressed payload comparison to compare slice references explicitly.
314
+- Changed bounded stream reads to clear the reusable buffer on both over-limit and read-error paths.
315
+- Added private small-cap helper entry points used by tests.
316
+- Added small custom `Read` tests for over-limit stream handling in both readers.
317
+
318
+Validation:
319
+
320
+- `cargo fmt -p journal_file -p journal_reader_ffi` in `src/crates/jf`: passed.
321
+- `cargo fmt -p journal-core` in `src/crates`: passed.
322
+- `cargo test -q` in `src/crates/jf`: passed; 7 tests passed.
323
+- `cargo test -q -p journal-core` in `src/crates`: passed; 21 tests passed plus the existing ignored tests.
324
+
325
+## PR Review Iteration 4 - 2026-05-08
326
+
327
+Findings:
328
+
329
+- `PRRT_kwDOAKPxd86AteMb`, `src/crates/journal-core/src/file/object.rs:900`: valid. The testable size-cap helper converted `usize` to `u64` with `as` and then added 1.
330
+- `PRRT_kwDOAKPxd86AteNJ`, `src/crates/jf/journal_file/src/file.rs:204`: valid readability issue. The `BucketVisitor` implementation used an elided matcher lifetime.
331
+- `PRRT_kwDOAKPxd86AteNi`, `src/crates/jf/journal_file/src/writer.rs:640`: valid. The corrected assertion message lost useful context.
332
+
333
+Actions:
334
+
335
+- Changed both bounded-read helpers to use `u64::try_from(max_size)` plus `checked_add(1)`.
336
+- Made `DataPayloadMatcher`'s borrowed payload lifetime explicit in the `BucketVisitor` implementation.
337
+- Restored context in the filter-count assertion message.
338
+
339
+Validation:
340
+
341
+- `cargo fmt -p journal_file -p journal_reader_ffi` in `src/crates/jf`: passed.
342
+- `cargo fmt -p journal-core` in `src/crates`: passed.
343
+- `cargo test -q` in `src/crates/jf`: passed; 7 tests passed.
344
+- `cargo test -q -p journal-core` in `src/crates`: passed; 21 tests passed plus the existing ignored tests.
345
+
346
+## PR Review Iteration 5 - 2026-05-08
347
+
348
+Findings:
349
+
350
+- `PRRT_kwDOAKPxd86Atn1r`, `src/crates/journal-core/src/file/object.rs:1047`: valid. The LZ4 decode-error path left the reusable buffer resized to the advertised uncompressed size.
351
+- `PRRT_kwDOAKPxd86Atn2C`, `src/crates/jf/journal_file/src/object.rs:941`: valid. Same LZ4 decode-error buffer state issue in the legacy static reader, including the FFI caller reuse path.
352
+
353
+Actions:
354
+
355
+- Changed both LZ4 decode-error branches to replace the reusable buffer with a new empty `Vec`, so callers cannot observe stale decompressed bytes or retain the failed allocation.
356
+- Added malformed LZ4 block regression tests for both readers.
357
+
358
+Validation:
359
+
360
+- `cargo fmt -p journal_file -p journal_reader_ffi` in `src/crates/jf`: passed.
361
+- `cargo fmt -p journal-core` in `src/crates`: passed.
362
+- `cargo test -q` in `src/crates/jf`: passed; 8 tests passed.
363
+- `cargo test -q -p journal-core` in `src/crates`: passed; 22 tests passed plus the existing ignored tests.
364
+- `git diff --check`: passed.
365
+
366
+## PR Review Iteration 6 - 2026-05-08
367
+
368
+Findings:
369
+
370
+- `PRRT_kwDOAKPxd86Atzu9`, `src/crates/journal-core/src/file/object.rs:916`: valid. The bounded Zstd/XZ stream helper cleared the buffer on over-limit or read errors but retained the potentially large allocation.
371
+- `PRRT_kwDOAKPxd86Atzv6`, `src/crates/jf/journal_file/src/object.rs:811`: valid. Same bounded stream allocation-retention issue in the legacy static reader.
372
+- `PRRT_kwDOAKPxd86Atzvg`, `src/crates/journal-core/src/file/object.rs:1052`: valid. The LZ4 path left `buf.len()` at the advertised size if the decoder returned fewer bytes than the prefix.
373
+- `PRRT_kwDOAKPxd86AtzwG`, `src/crates/jf/journal_file/src/object.rs:946`: valid. Same LZ4 size/length mismatch semantics in the legacy static reader.
374
+
375
+Actions:
376
+
377
+- Changed both bounded stream helpers to replace the reusable buffer with a new empty `Vec` on over-limit or read errors.
378
+- Changed both LZ4 branches to accept success only when the decoded length matches the systemd uncompressed-size prefix; mismatch now resets the buffer and returns `JournalError::DecompressorError`.
379
+- Added LZ4 size-mismatch regression tests in both readers and strengthened bounded stream tests to assert capacity release.
380
+
381
+Validation:
382
+
383
+- `cargo fmt -p journal_file -p journal_reader_ffi` in `src/crates/jf`: passed.
384
+- `cargo fmt -p journal-core` in `src/crates`: passed.
385
+- `cargo test -q` in `src/crates/jf`: passed; 9 tests passed.
386
+- `cargo test -q -p journal-core` in `src/crates`: passed; 23 tests passed plus the existing ignored tests.
387
+- `git diff --check`: passed.
388
+
389
+## PR Review Iteration 7 - 2026-05-08
390
+
391
+Findings:
392
+
393
+- `PRRT_kwDOAKPxd86At82Y`, `src/crates/journal-core/src/file/object.rs:915`: valid. `read_to_end()` with `take(max + 1)` bounded decompressed bytes but still allowed `Vec` growth strategy to over-allocate beyond the intended cap.
394
+- `PRRT_kwDOAKPxd86At821`, `src/crates/jf/journal_file/src/file.rs:583`: valid. Existing tests covered `DataPayloadMatcher::payload_matches()` directly but not the `find_data_offset()` hash-bucket traversal path that uses the matcher.
395
+
396
+Actions:
397
+
398
+- Replaced bounded streaming `read_to_end()` calls in both readers with a manual fixed-size stack-buffer loop that uses `try_reserve_exact()` for each chunk and fails as soon as the decompressed stream exceeds the configured cap.
399
+- Added a temporary-journal test that writes a compressed DATA object into the data hash table and verifies `find_data_offset()` locates it by logical uncompressed payload.
400
+
401
+Validation:
402
+
403
+- `cargo fmt -p journal_file -p journal_reader_ffi` in `src/crates/jf`: passed.
404
+- `cargo fmt -p journal-core` in `src/crates`: passed.
405
+- `cargo test -q` in `src/crates/jf`: passed; 10 tests passed.
406
+- `cargo test -q -p journal-core` in `src/crates`: passed; 23 tests passed plus the existing ignored tests.
407
+- `git diff --check`: passed.
408
+
409
+## PR Review Iteration 8 - 2026-05-08
410
+
411
+Findings:
412
+
413
+- `PRRT_kwDOAKPxd86AuG7E`, `src/crates/journal-core/src/file/object.rs:926`: valid. Per-read `try_reserve_exact()` avoided unbounded growth but could cause one allocation per read chunk for large decompressed payloads.
414
+- `PRRT_kwDOAKPxd86AuG7c`, `src/crates/jf/journal_file/src/file.rs:1190`: reviewed and not changed. The direct matcher fixture's `ObjectHeader::size = header + payload.len()` matches the writer contract; object placement alignment is handled by `ObjectHeader::aligned_size()` and is covered by the new temporary-journal `find_data_offset()` test.
415
+
416
+Actions:
417
+
418
+- Changed both bounded stream helpers to reserve capacity in amortized chunks, growing up to the configured decompressed-size cap without using `read_to_end()`'s geometric growth.
419
+- Kept the direct matcher helper unpadded because padding those raw bytes would become part of the payload passed to `DataObject::from_data()` and would diverge from the writer's unpadded `size` field.
420
+
421
+Validation:
422
+
423
+- `cargo fmt -p journal_file -p journal_reader_ffi` in `src/crates/jf`: passed.
424
+- `cargo fmt -p journal-core` in `src/crates`: passed.
425
+- `cargo test -q` in `src/crates/jf`: passed; 10 tests passed.
426
+- `cargo test -q -p journal-core` in `src/crates`: passed; 23 tests passed plus the existing ignored tests.
427
+- `git diff --check`: passed.
428
+
429
+## PR Review Iteration 9 - 2026-05-08
430
+
431
+Findings:
432
+
433
+- `PRRT_kwDOAKPxd86AuPUu`, `src/crates/journal-core/src/file/object.rs:947`: valid. The amortized `try_reserve_exact()` call still used a capacity delta, but the API takes additional capacity relative to `buf.len()`.
434
+- `PRRT_kwDOAKPxd86AuPVH`, `src/crates/jf/journal_file/src/object.rs:842`: valid. Same reservation argument bug in the legacy reader.
435
+
436
+Actions:
437
+
438
+- Changed both amortized reservation paths to pass `target_capacity - buf.len()` to `try_reserve_exact()`.
439
+
440
+Validation:
441
+
442
+- `cargo fmt -p journal_file -p journal_reader_ffi` in `src/crates/jf`: passed.
443
+- `cargo fmt -p journal-core` in `src/crates`: passed.
444
+- `cargo test -q` in `src/crates/jf`: passed; 10 tests passed.
445
+- `cargo test -q -p journal-core` in `src/crates`: passed; 23 tests passed plus the existing ignored tests.
446
+- `git diff --check`: passed.
447
+
448
+## Outcome
449
+
450
+Implemented, validated, and prepared for commit in `~/src/PRs/netdata-static-journal-facets`.
451
+
452
+## Lessons Extracted
453
+
454
+- The legacy `src/crates/jf/` reader and the newer `src/crates/journal-core/` reader had diverged on compression support. For journal-format fixes, compare both implementations before changing one path.
455
+
456
+## Followup
457
+
458
+None for the completed PR review iteration; the appended regression section records the live validation repair.
459
+
460
+## Regression Log
461
+
462
+## Regression - 2026-05-08
463
+
464
+What broke:
465
+
466
+- Live static install testing on `[LOCAL_RHEL_8_10_TEST_AGENT]:19999` still returned empty rows when a systemd-journal facet value was selected.
467
+- The previous validation only proved compressed DATA lookup/decompression at the Rust crate level. It did not prove the full C plugin Function path that receives facet selections from the UI, converts them into backend matches, and streams rows back to the dashboard.
468
+
469
+Evidence:
470
+
471
+- User report: static build copied and installed on a local RHEL 8.10 test Agent; selecting a facet returns empty responses.
472
+- The regression is specific to facet filtering, not unfiltered journal access.
473
+- Direct agent Function evidence from `[LOCAL_RHEL_8_10_TEST_AGENT]:19999`: unfiltered `systemd-journal` with `slice:true` returned 20 rows and non-empty facets for the last 4 hours; selecting advertised values `SYSLOG_IDENTIFIER=netdata`, `_SYSTEMD_UNIT=netdata.service`, or `PRIORITY=6` returned 0 rows with `slice:true`, while the same selections returned 20 rows with `slice:false`.
474
+- Direct response stats for `SYSLOG_IDENTIFIER=netdata`, `slice:true`: request echoes the expected JSON `selections`, but `rows.evaluated=0`, `rows.matched=0`, and the journal file reports `rows_read=0`. This matches a failure before row scanning, during native match setup / filtered cursor resolution.
475
+- Source evidence: `src/crates/jf/journal_file/src/filter.rs:339-357` builds filtered cursors by hashing selected `field=value` bytes and resolving them through `find_data_offset()`.
476
+- Source evidence: `src/crates/jf/journal_file/src/hash.rs:4-8` still uses `twox_hash::XxHash64` behind a `FIXME` for the non-keyed Jenkins path.
477
+- Source evidence: `systemd/systemd @ d0c912899a33436d6676b2564eb1ac506f378571`, `src/libsystemd/sd-journal/journal-file.c:1585-1600`, uses `siphash24()` only for keyed journal files and `jenkins_hash64()` otherwise.
478
+- Source evidence: `systemd/systemd @ d0c912899a33436d6676b2564eb1ac506f378571`, `src/libsystemd/sd-journal/lookup3.h:14-20`, defines `jenkins_hash64()` as lookup3 `jenkins_hashlittle2()` with the primary value in the high 32 bits and the secondary value in the low 32 bits.
479
+- In-repository pattern: `src/crates/journal-core/src/file/hash.rs:4-17` already uses `hashers::jenkins::Lookup3Hasher` and swaps the 32-bit halves to match systemd's `jenkins_hash64()`.
480
+
481
+Repair plan:
482
+
483
+- Replace the legacy `src/crates/jf` non-keyed hash fallback with the same lookup3/Jenkins implementation used by `journal-core`.
484
+- Add reference-value tests for systemd-compatible Jenkins hashes.
485
+- Re-run focused Rust tests and direct selected-facet queries on the local RHEL 8.10 static install.
486
+
487
+Validation plan:
488
+
489
+- Focused Rust crate tests for the fixed path.
490
+- Direct API evidence from the local RHEL 8.10 static install showing the same facet values return rows after the fix.
491
+- Static build/install validation using the user-provided local static-binary workflow if a code change is needed.
492
+
493
+Why previous validation missed it:
494
+
495
+- The first fix validated compressed DATA enumeration and compressed payload lookup, but did not validate the full selected-facet Function path on a real non-keyed journal file.
496
+- The failing `slice:true` path builds filtered cursors before scanning rows; the incorrect hash function made `find_data_offset()` fail before row evaluation, so decompression tests alone could not catch it.
497
+
498
+Actions:
499
+
500
+- Replaced the legacy `src/crates/jf` non-keyed journal hash implementation with `hashers::jenkins::Lookup3Hasher`, matching the existing `journal-core` pattern and systemd lookup3 half ordering.
501
+- Added the empty-payload lookup3 guard because `hashers` returns `0` for empty input while systemd lookup3 with zero seeds returns `0xdeadbeefdeadbeef`.
502
+- Added systemd reference-value tests for the legacy reader hash path.
503
+- Applied the same empty-payload guard and reference-value test to `src/crates/journal-core` so the legacy and newer journal readers remain consistent.
504
+
505
+Validation:
506
+
507
+- `cargo fmt -p journal_file -p journal_reader_ffi` in `src/crates/jf`: passed.
508
+- `cargo fmt -p journal-core` in `src/crates`: passed.
509
+- `cargo test -q` in `src/crates/jf`: passed; 11 tests passed.
510
+- `cargo test -q -p journal-core` in `src/crates`: passed; 24 tests passed plus the existing ignored tests.
511
+- Static build via the user-provided local static-binary workflow: passed; produced `artifacts/netdata-x86_64-latest.gz.run`.
512
+- Static install on the local RHEL 8.10 test Agent: passed; installer restarted `netdata` through systemd.
513
+- Direct Function validation after install:
514
+ - baseline `slice:true`: `data_len=20`, `rows_evaluated=2019`, `rows_matched=2019`, `rows_read=2019`.
515
+ - `SYSLOG_IDENTIFIER=netdata`, `slice:true`: `data_len=20`, `rows_evaluated=1499`, `rows_matched=1499`, `rows_read=1499`.
516
+ - `_SYSTEMD_UNIT=netdata.service`, `slice:true`: `data_len=20`, `rows_evaluated=1714`, `rows_matched=1714`, `rows_read=1714`.
517
+ - `PRIORITY=6`, `slice:true`: `data_len=20`, `rows_evaluated=1596`, `rows_matched=1596`, `rows_read=1596`.
518
+ - `SYSLOG_IDENTIFIER=netdata`, `slice:false`: `data_len=20`, `rows_evaluated=2019`, `rows_matched=1499`, `rows_read=2019`.
519
+- Same-failure scan: `rg -n "twox_hash|XxHash64|jenkins_hash|Lookup3Hasher|hashers" src/crates/jf src/crates/journal-core src/crates/Cargo.toml src/crates/Cargo.lock` found no remaining `twox_hash`/`XxHash64` use in the legacy journal hash path, and found both journal readers using `Lookup3Hasher`.
520
+- `git diff --check`: passed.
521
+- `.agents/sow/audit.sh`: SOW status/directory checks passed for this SOW; audit still exits non-zero for the pre-existing public SSH clone syntax false positive in `.agents/skills/mirror-netdata-repos/SKILL.md:112`, unrelated to this work.
522
+
523
+Sensitive data handling:
524
+
525
+- Bearer tokens, Cloud token values, claim identifiers, node identifiers, raw journal rows, and private endpoint names were not written to this SOW.
526
+- Raw response JSON from live validation was kept under `.local/audits/`, which is gitignored and not staged.
527
+
528
+Artifact updates:
529
+
530
+- AGENTS.md: no update needed; workflow and project guardrails did not change.
531
+- Runtime project skills: no update needed; the static-build skill worked as operational context and the fix did not change how agents should work on collectors.
532
+- Specs: no update needed; this restores the intended systemd-compatible journal hash behavior.
533
+- End-user/operator docs: no update needed; no user-facing configuration, command, or workflow changed.
534
+- End-user/operator skills: no update needed; public/operator skill behavior was not changed by this PR.
535
+- SOW lifecycle: SOW is moved back to `.agents/sow/done/` with `Status: completed` in the same commit as the regression repair.
536
+
537
+Follow-up mapping:
538
+
539
+- No follow-up remains for the static journal facet filtering regression.
540
+
541
+## Core Library Extension - 2026-05-09
542
+
543
+Question answered:
544
+
545
+- The user asked whether the same compatibility-layer changes were also needed in the underlying `journal-core` library.
546
+
547
+Evidence:
548
+
549
+- `src/crates/journal-core/src/file/hash.rs:4-64` already had the systemd-compatible Jenkins lookup3 hash implementation and reference-value test from the regression repair.
550
+- `src/crates/journal-core/src/file/object.rs:1049-1103` already had bounded Zstd/LZ4/XZ DATA decompression support from the earlier PR review iterations.
551
+- `src/crates/journal-core/src/file/file.rs:39-79` still used the generic `PayloadMatcher`, which compared `object.raw_payload() == self.payload`.
552
+- `src/crates/journal-core/src/file/file.rs:485-488` used that raw-only matcher in `find_data_offset()`.
553
+- `src/crates/journal-core/src/file/filter.rs:283` and `src/crates/journal-core/src/file/filter.rs:297` build filter expressions through `find_data_offset()`, so compressed DATA objects could still fail filter construction in core-library users.
554
+
555
+Actions:
556
+
557
+- Added `DataPayloadMatcher` to `src/crates/journal-core/src/file/file.rs`, matching the `src/crates/jf` repair pattern.
558
+- Kept the raw-payload comparison as the first path for uncompressed DATA objects.
559
+- Added a compressed-payload comparison path that decompresses only compressed DATA objects in the selected hash bucket.
560
+- Changed `journal-core` `find_data_offset()` to use `DataPayloadMatcher`.
561
+- Added focused `journal-core` tests for direct LZ4 compressed matching, negative compressed matching, and the `find_data_offset()` hash-bucket traversal path.
562
+
563
+Validation:
564
+
565
+- `cargo fmt -p journal-core` in `src/crates`: passed.
566
+- `cargo test -q -p journal-core` in `src/crates`: passed; 27 tests passed plus the existing ignored tests.
567
+- `cargo test -q` in `src/crates/jf`: passed; 11 tests passed.
568
+- Same-failure scan: `rg -n "raw_payload\\(\\) == self\\.payload|DataPayloadMatcher|find_data_offset\\(|payload_matches" src/crates/jf src/crates/journal-core` confirmed both readers now use `DataPayloadMatcher` for DATA hash-bucket lookup; the remaining raw-payload comparison is the generic field-object matcher path, which is correct because FIELD objects are not compressed DATA payloads.
569
+
570
+Artifact updates:
571
+
572
+- AGENTS.md: no update needed; workflow and project guardrails did not change.
573
+- Runtime project skills: no update needed; this is a code-path consistency fix, not a workflow change.
574
+- Specs: no update needed; this preserves the intended journal-format behavior.
575
+- End-user/operator docs: no update needed; no user-facing configuration, command, or workflow changed.
576
+- End-user/operator skills: no update needed; public/operator skill behavior was not changed.
577
+- SOW lifecycle: same SOW reopened for the same-class core-library gap and moved back to `.agents/sow/done/` with `Status: completed` in the same commit.
578
+
579
+Follow-up mapping:
580
+
581
+- No follow-up remains for the underlying `journal-core` compressed DATA lookup gap.
582
+
583
+## External Review Follow-up - 2026-05-09
584
+
585
+Why reopened:
586
+
587
+- The user requested external review of the performance and side effects of the changes before merge.
588
+- Reviewers agreed the original static facet fix is correct, but flagged same-family raw-payload reads outside the original `find_data_offset()` lookup path.
589
+
590
+Confirmed findings:
591
+
592
+- `src/crates/journal-core/src/file/file.rs:595` read remapping entry DATA bytes with `raw_payload()` after the marker lookup had become compressed-aware.
593
+- `src/crates/journal-core/src/file/reader.rs:428` copied remapping entry DATA bytes with `raw_payload()`.
594
+- `src/crates/journal-index/src/field_types.rs:233` parsed source timestamp DATA bytes with `raw_payload()`, with callers in `src/crates/journal-index/src/file_index.rs:390` and `src/crates/journal-index/src/file_indexer.rs:437`.
595
+- The `hashers` crate contains suspicious optimized alignment branches, but local inspection showed its `offset_to_align()` helper never returns `0` for normal alignments, so the crate falls back to the byte path used by the passing reference-value tests and the live RHEL 8.10 validation. This is not a current PR blocker.
596
+
597
+Actions:
598
+
599
+- Added `DataObject::logical_payload()` in `src/crates/journal-core/src/file/object.rs`.
600
+- Updated `journal-core` remapping reads to use logical payload bytes and return `JournalError::InvalidField` instead of panicking on non-UTF-8 data.
601
+- Updated `journal-index` source timestamp parsing to use logical payload bytes, while preserving a scratch-buffer variant to avoid repeated allocations in loops.
602
+- Added focused `journal-core` tests for logical raw payloads and LZ4-compressed logical payloads.
603
+
604
+Validation:
605
+
606
+- `cargo fmt -p journal-core -p journal-index` in `src/crates`: passed.
607
+- `cargo test -q -p journal-core` in `src/crates`: passed; 29 tests passed plus the existing ignored doc tests.
608
+- `cargo test -q -p journal-index` in `src/crates`: passed; 66 tests passed across the package test binaries.
609
+
610
+Follow-up mapping:
611
+
612
+- Stale uncompiled `src/crates/jf/journal_file/src/journal_file.rs` still contains a raw-payload lookup helper with no callers. This is not runtime behavior for this PR and should be handled only if that stale module is removed or revived.
613
+- Zstd/XZ integration tests through `find_data_offset()` would improve coverage but do not block this RHEL LZ4 regression fix because object-level XZ and bounded streaming behavior are already tested, and the logical matcher delegates to the same decompression API.
614
+
615
+## External Review Rerun Closure - 2026-05-09
616
+
617
+Why updated:
618
+
619
+- The user asked to run the external reviewers again after the first follow-up fixes.
620
+- The rerun found no evidence that the live RHEL 8.10 static facet failure remained, but it did identify two local hardening issues in the same code path.
621
+
622
+Confirmed findings:
623
+
624
+- `src/crates/journal-core/src/file/file.rs:629` still used `expect("utf8 data")` for FIELD names in the same `load_fields()` path where DATA payload parsing had already been converted to `JournalError::InvalidField`.
625
+- `src/collectors/systemd-journal.plugin/provider/rust_provider.h:13` exposed unique enumeration through a foreach macro that stops on `<= 0`; when `rsd_journal_enumerate_available_unique()` returned a negative decompression error, `src/collectors/systemd-journal.plugin/systemd-journal.c:583` could silently stop without incrementing `failures`, so the fallback at `systemd-journal.c:621` would not run.
626
+- `src/crates/journal-index/src/file_index.rs:493` manually decompressed regex payloads even though `DataObject::logical_payload()` now exists.
627
+
628
+Reviewed and rejected as non-blocking for this SOW:
629
+
630
+- `src/crates/journal-index/src/file_indexer.rs:39-43` documents that compressed values are skipped by the bitmap indexer; `file_indexer.rs:296-309` implements that existing index-size policy. Changing it would require a separate product/performance decision because it would decompress and index every unique compressed value. This PR keeps the documented indexing limit unchanged.
631
+- A reviewer claimed the Zstd object flag should be `1 << 3`. That finding was false. Current systemd source has `OBJECT_COMPRESSED_ZSTD = 1 << 2` and `HEADER_INCOMPATIBLE_COMPRESSED_ZSTD = 1 << 3`; Netdata's object/header constants match that split. Evidence: `https://raw.githubusercontent.com/systemd/systemd/main/src/libsystemd/sd-journal/journal-def.h`, lines 62-64 and 186.
632
+- The stale uncompiled `src/crates/jf/journal_file/src/journal_file.rs` helper still has no runtime callers. It is rejected for this SOW because changing dead code would increase review surface without changing shipped behavior.
633
+- Zstd/XZ `find_data_offset()` integration tests and Zstd object-level tests are useful coverage, but they are rejected for this SOW because the production regression is the RHEL LZ4 path and the shared decompression paths are already covered by object-level LZ4/XZ and bounded-stream tests.
634
+
635
+Actions:
636
+
637
+- Converted the remaining FIELD-name panic in `journal-core` `load_fields()` to `JournalError::InvalidField`.
638
+- Added `nsd_journal_enumerate_available_unique()` to the provider abstraction.
639
+- Replaced the filter builder's `NSD_JOURNAL_FOREACH_UNIQUE()` macro use with an explicit restart/enumerate loop that counts negative `query_unique()` and enumeration returns as setup failures, preserving the existing full-query fallback.
640
+- Replaced `_BOOT_ID` annotation unique enumeration with the same explicit wrapper and logged negative enumeration returns.
641
+- Replaced the manual regex-path decompression in `journal-index` with `DataObject::logical_payload()`.
642
+
643
+Validation:
644
+
645
+- `curl -fsSL https://raw.githubusercontent.com/systemd/systemd/main/src/libsystemd/sd-journal/journal-def.h | rg -n "OBJECT_COMPRESSED_(XZ|LZ4|ZSTD)|HEADER_INCOMPATIBLE_COMPRESSED_ZSTD"`: verified object Zstd is `1 << 2` and header incompatible Zstd is `1 << 3`.
646
+- `cargo fmt -p journal-core -p journal-index` in `src/crates`: passed.
647
+- `git diff --check`: passed.
648
+- `cargo test -q -p journal-core` in `src/crates`: passed; 29 tests passed plus the existing ignored doc tests.
649
+- `cargo test -q -p journal-index` in `src/crates`: passed; 66 tests passed across the package test binaries.
650
+- `cargo test -q --all-targets` in `src/crates/jf`: passed; 11 tests passed.
651
+- `./packaging/makeself/build-static.sh x86_64`: passed; produced `artifacts/netdata-x86_64-latest.gz.run` after compiling `systemd-journal.plugin`, `journal_reader_ffi`, `journal-core`, and `journal-index` in the static musl build.
652
+- `.agents/sow/audit.sh`: SOW status/directory checks passed for this SOW; the audit still exits non-zero on the pre-existing public SSH clone syntax pattern in `.agents/skills/mirror-netdata-repos/SKILL.md:112`, unrelated to this work and not staged by this PR.
653
+
654
+Follow-up mapping:
655
+
656
+- No follow-up remains for the static journal facet filtering regression.
657
+
658
+## Libsystemd Compatibility Follow-up - 2026-05-09
659
+
660
+Why reopened:
661
+
662
+- The user asked whether the new explicit C call could break old `libsystemd` builds if older `SD_JOURNAL_FOREACH_UNIQUE()` macros did not use `sd_journal_enumerate_available_unique()`.
663
+
664
+Evidence:
665
+
666
+- systemd v245 `src/systemd/sd-journal.h` declares `sd_journal_enumerate_unique()` and defines `SD_JOURNAL_FOREACH_UNIQUE()` with `sd_journal_enumerate_unique()`.
667
+- systemd v246 `src/systemd/sd-journal.h` declares `sd_journal_enumerate_available_unique()` and defines `SD_JOURNAL_FOREACH_UNIQUE()` with `sd_journal_enumerate_available_unique()`.
668
+- The current systemd manual records `sd_journal_query_unique()`, `sd_journal_enumerate_unique()`, `sd_journal_restart_unique()`, and `SD_JOURNAL_FOREACH_UNIQUE()` as added in version 195; `sd_journal_enumerate_available_unique()` was added in version 246.
669
+
670
+Conclusion:
671
+
672
+- The user's concern was valid. The previous C wrapper would have broken builds against libsystemd headers older than v246 when `HAVE_SD_JOURNAL_RESTART_FIELDS` was set.
673
+
674
+Actions:
675
+
676
+- Added `HAVE_SD_JOURNAL_ENUMERATE_AVAILABLE_UNIQUE` detection in `packaging/cmake/Modules/NetdataDetectSystemd.cmake`.
677
+- Added the generated config define to `packaging/cmake/config.cmake.h.in`.
678
+- Updated `nsd_journal_enumerate_available_unique()` so:
679
+ - Rust provider builds call `rsd_journal_enumerate_available_unique()`.
680
+ - Modern libsystemd builds call `sd_journal_enumerate_available_unique()`.
681
+ - Older libsystemd builds fall back to `sd_journal_enumerate_unique()`, matching the old `SD_JOURNAL_FOREACH_UNIQUE()` macro behavior.
682
+
683
+Validation:
684
+
685
+- systemd v245 header check: confirmed `SD_JOURNAL_FOREACH_UNIQUE()` uses `sd_journal_enumerate_unique()`.
686
+- systemd v246 header check: confirmed `SD_JOURNAL_FOREACH_UNIQUE()` uses `sd_journal_enumerate_available_unique()`.
687
+- Current systemd manual check: confirmed `sd_journal_enumerate_available_unique()` was added in version 246.
688
+- Compiled `src/collectors/systemd-journal.plugin/provider/netdata_provider.c` with a temporary config where `HAVE_SD_JOURNAL_RESTART_FIELDS` is defined and `HAVE_SD_JOURNAL_ENUMERATE_AVAILABLE_UNIQUE` is not defined: passed, proving the old-libsystemd branch compiles.
689
+- Compiled the same provider file with both `HAVE_SD_JOURNAL_RESTART_FIELDS` and `HAVE_SD_JOURNAL_ENUMERATE_AVAILABLE_UNIQUE` defined: passed, proving the modern-libsystemd branch compiles.
690
+- `git diff --check`: passed.
691
+
692
+Artifact updates:
693
+
694
+- AGENTS.md: no update needed; workflow and project guardrails did not change.
695
+- Runtime project skills: no update needed; this is a compatibility guard in project code.
696
+- Specs: no update needed; this preserves existing libsystemd compatibility.
697
+- End-user/operator docs: no update needed; no user-facing configuration, command, or workflow changed.
698
+- End-user/operator skills: no update needed; public/operator skill behavior was not changed.
699
+- SOW lifecycle: SOW reopened for this compatibility fix and will be moved back to done with `Status: completed` in the same commit.
700
+
701
+Follow-up mapping:
702
+
703
+- No follow-up remains for libsystemd unique-enumeration symbol compatibility.
704
+
705
+## PR Review Cleanup - 2026-05-09
706
+
707
+Why reopened:
708
+
709
+- The user reported that old PR comments/reviews were still unresolved.
710
+- Current PR review state showed three unresolved bot review threads.
711
+
712
+Findings:
713
+
714
+- `PRRT_kwDOAKPxd86AyMvL`, `src/crates/journal-core/src/file/object.rs:1085`: valid. Several `DataObject::decompress()` error paths can return without clearing the caller-provided scratch buffer, including short LZ4 prefix, oversized LZ4 prefix, LZ4 reserve failure, Zstd decoder creation failure, and unknown compression method.
715
+- `PRRT_kwDOAKPxd86AyMwE`, `src/crates/jf/journal_file/src/object.rs:971`: valid. The legacy static reader has the same scratch-buffer error-path issue.
716
+- `PRRT_kwDOAKPxd86Aye5J`, `src/crates/journal-index/src/field_types.rs:242`: valid. `get_timestamp_field()` scans all entry DATA objects for the configured timestamp field. A decompression failure in an unrelated compressed DATA object can abort the search instead of being treated like a non-match and allowing `get_entry_timestamp()` to fall back to the entry realtime timestamp.
717
+
718
+Planned actions:
719
+
720
+- Reset decompression scratch buffers to a new empty `Vec` on every `DataObject::decompress()` error path that occurs before the bounded stream reader or LZ4 decoder already clears it.
721
+- Add regression tests in both readers for short LZ4 prefixes and stale-buffer oversized prefixes.
722
+- Treat compressed-payload decompression failures as timestamp-field non-matches during timestamp parsing, while preserving other journal errors.
723
+- Run focused Rust validation and GitHub review sync before commit/push.
724
+
725
+Actions:
726
+
727
+- Changed `src/crates/journal-core/src/file/object.rs` and `src/crates/jf/journal_file/src/object.rs` so early decompression failures reset the scratch buffer to a new empty `Vec`.
728
+- Added short LZ4-prefix regression tests in both readers.
729
+- Strengthened the oversized LZ4-prefix tests in both readers to start with a stale buffer and assert capacity release.
730
+- Changed `src/crates/journal-index/src/field_types.rs` so `JournalError::DecompressorError` and `JournalError::UnknownCompressionMethod` become `IndexError::InvalidFieldPrefix` for timestamp parsing. Other journal errors still propagate.
731
+- Added a focused timestamp error-classification test.
732
+
733
+Validation:
734
+
735
+- `cargo fmt -p journal-core -p journal-index` in `src/crates`: passed.
736
+- `cargo fmt -p journal_file -p journal_reader_ffi` in `src/crates/jf`: passed.
737
+- `cargo test -q -p journal-core` in `src/crates`: passed; 30 tests passed plus existing ignored doc tests.
738
+- `cargo test -q -p journal-index` in `src/crates`: passed; 67 tests passed across package test binaries.
739
+- `cargo test -q --all-targets` in `src/crates/jf`: passed; 12 tests passed.
740
+- `git diff --check`: passed.
741
+- Same-failure scan: checked the affected decompression error returns and confirmed the remaining early returns in both `DataObject::decompress()` implementations clear the scratch buffer before returning; bounded-stream and LZ4 decoder failure paths already clear internally.
742
+- PR sync barrier: `fetch-all.sh 22456` still showed the same three unresolved threads and no newer unresolved threads.
743
+- Sonar sync barrier: `fetch-sonar-findings.sh 22456` reported 0 issues and 0 hotspots.
744
+- CI sync barrier: `ci-status.sh 22456` reported 0 failing checks and 94 running checks before this push.
745
+- `.agents/sow/audit.sh`: SOW status/directory checks passed; audit still exits non-zero on the pre-existing public SSH clone syntax pattern in `.agents/skills/mirror-netdata-repos/SKILL.md:112`, unrelated to this work and not staged by this PR.
746
+
747
+Artifact updates:
748
+
749
+- AGENTS.md: no update needed; workflow and project guardrails did not change.
750
+- Runtime project skills: no update needed; this is a code review cleanup in an existing workflow.
751
+- Specs: no update needed; this preserves intended journal error resilience.
752
+- End-user/operator docs: no update needed; no user-facing configuration, command, or workflow changed.
753
+- End-user/operator skills: no update needed; public/operator skill behavior was not changed.
754
+- SOW lifecycle: SOW reopened for unresolved PR comments and will be moved back to done with `Status: completed` in the same commit.
755
+
756
+Follow-up mapping:
757
+
758
+- No follow-up remains for these unresolved review comments.
759
+
760
+## PR Review Rerun Cleanup - 2026-05-09
761
+
762
+Why reopened:
763
+
764
+- Re-triggered PR review after commit `709857f261` opened three new bot review threads.
765
+
766
+Findings:
767
+
768
+- `PRRT_kwDOAKPxd86Aymui`, `src/crates/journal-core/src/file/file.rs:67`: valid. `DataPayloadMatcher::payload_matches()` propagated decompression failures, so one corrupt compressed DATA object in a hash bucket could stop lookup before a later matching object.
769
+- `PRRT_kwDOAKPxd86Aymuv`, `src/crates/jf/journal_file/src/file.rs:184`: valid. The legacy static reader had the same hash-bucket lookup resilience issue.
770
+- `PRRT_kwDOAKPxd86Aymu0`, `src/collectors/systemd-journal.plugin/systemd-journal-files.c:173`: valid. The log message named `sd_journal_enumerate_available_unique()` even though the call goes through `nsd_journal_enumerate_available_unique()`, which can fall back to older libsystemd APIs.
771
+
772
+Planned actions:
773
+
774
+- Treat DATA-object decompression failures in both payload matchers as non-matches so hash-bucket traversal continues.
775
+- Add bucket-level tests where a bad compressed DATA object precedes a valid matching compressed DATA object in the same bucket.
776
+- Update the C diagnostic to name the provider wrapper.
777
+- Re-run focused tests and PR sync before commit/push.
778
+
779
+Actions:
780
+
781
+- Changed both `DataPayloadMatcher::payload_matches()` implementations to return `Ok(false)` for `JournalError::DecompressorError` and `JournalError::UnknownCompressionMethod`.
782
+- Added bucket-level regression tests in both readers proving `find_data_offset()` skips a bad compressed DATA object and finds a later valid compressed DATA object in the same hash bucket.
783
+- Changed the `_BOOT_ID` annotation diagnostic to name `nsd_journal_enumerate_available_unique()`.
784
+
785
+Validation:
786
+
787
+- `cargo fmt -p journal-core` in `src/crates`: passed.
788
+- `cargo fmt -p journal_file -p journal_reader_ffi` in `src/crates/jf`: passed.
789
+- `cargo test -q -p journal-core` in `src/crates`: passed; 31 tests passed plus existing ignored doc tests.
790
+- `cargo test -q --all-targets` in `src/crates/jf`: passed; 13 tests passed.
791
+- `git diff --check`: passed.
792
+- Same-failure scan: `rg` found no remaining `decompress(&mut self.decompressed_payload)?` direct propagation and confirmed the diagnostic now names the wrapper.
793
+- PR sync barrier: `fetch-all.sh 22456` still showed the same three unresolved rerun threads and no newer unresolved threads.
794
+- Sonar sync barrier: `fetch-sonar-findings.sh 22456` reported 0 issues and 0 hotspots.
795
+- CI sync barrier: `ci-status.sh 22456` reported 0 failing checks and 94 running checks before this push.
796
+- `.agents/sow/audit.sh`: SOW status/directory checks passed; audit still exits non-zero on the pre-existing public SSH clone syntax pattern in `.agents/skills/mirror-netdata-repos/SKILL.md:112`, unrelated to this work and not staged by this PR.
797
+
798
+Artifact updates:
799
+
800
+- AGENTS.md: no update needed; workflow and project guardrails did not change.
801
+- Runtime project skills: no update needed; this is continued PR review cleanup.
802
+- Specs: no update needed; this preserves intended journal lookup resilience.
803
+- End-user/operator docs: no update needed; no user-facing configuration, command, or workflow changed.
804
+- End-user/operator skills: no update needed; public/operator skill behavior was not changed.
805
+- SOW lifecycle: SOW reopened for rerun review comments and moved back to done with `Status: completed` in the same commit.
806
+
807
+Follow-up mapping:
808
+
809
+- No follow-up remains for these rerun review comments.
packaging/cmake/Modules/NetdataDetectSystemd.cmake
+1
@@ -24,6 +24,7 @@ macro(detect_systemd)
24
check_symbol_exists(SD_JOURNAL_OS_ROOT "systemd/sd-journal.h" HAVE_SD_JOURNAL_OS_ROOT)
25
check_symbol_exists(sd_journal_open_files_fd "systemd/sd-journal.h" HAVE_SD_JOURNAL_OPEN_FILES_FD)
26
check_symbol_exists(sd_journal_restart_fields "systemd/sd-journal.h" HAVE_SD_JOURNAL_RESTART_FIELDS)
27
+ check_symbol_exists(sd_journal_enumerate_available_unique "systemd/sd-journal.h" HAVE_SD_JOURNAL_ENUMERATE_AVAILABLE_UNIQUE)
28
check_symbol_exists(sd_journal_get_seqnum "systemd/sd-journal.h" HAVE_SD_JOURNAL_GET_SEQNUM)
29
30
check_symbol_exists(sd_bus_default_system "systemd/sd-bus.h" HAVE_SD_BUS_DEFAULT_SYSTEM)
packaging/cmake/config.cmake.h.in
+1
@@ -103,6 +103,7 @@
103
#cmakedefine HAVE_SD_JOURNAL_OS_ROOT
104
#cmakedefine HAVE_SD_JOURNAL_OPEN_FILES_FD
105
#cmakedefine HAVE_SD_JOURNAL_RESTART_FIELDS
106
+#cmakedefine HAVE_SD_JOURNAL_ENUMERATE_AVAILABLE_UNIQUE
107
#cmakedefine HAVE_SD_JOURNAL_GET_SEQNUM
108
#cmakedefine ENABLE_SYSTEMD_DBUS
109
src/collectors/systemd-journal.plugin/provider/netdata_provider.c
+18
@@ -145,6 +145,24 @@ void nsd_journal_restart_unique(NsdJournal *j)
145
}
146
#endif /* HAVE_SD_JOURNAL_RESTART_FIELDS */
147
148
+#if defined(HAVE_SD_JOURNAL_RESTART_FIELDS)
149
+int nsd_journal_enumerate_available_unique(NsdJournal *j, const void **data, size_t *l)
150
+{
151
+#if defined(HAVE_RUST_PROVIDER)
152
+ uintptr_t rust_size = 0;
153
+ int r = rsd_journal_enumerate_available_unique(j, data, &rust_size);
154
+ if (r > 0)
155
+ *l = (size_t) rust_size;
156
+
157
+ return r;
158
+#elif defined(HAVE_SD_JOURNAL_ENUMERATE_AVAILABLE_UNIQUE)
159
+ return sd_journal_enumerate_available_unique(j, data, l);
160
+#else
161
+ return sd_journal_enumerate_unique(j, data, l);
162
+#endif
163
+}
164
+#endif /* HAVE_SD_JOURNAL_RESTART_FIELDS */
165
+
166
int nsd_journal_add_match(NsdJournal *j, const void *data, uintptr_t size)
167
{
168
#if defined(HAVE_RUST_PROVIDER)
src/collectors/systemd-journal.plugin/provider/netdata_provider.h
+1
@@ -64,6 +64,7 @@ void nsd_journal_restart_fields(NsdJournal *j);
64
65
int nsd_journal_query_unique(NsdJournal *j, const char *field);
66
void nsd_journal_restart_unique(NsdJournal *j);
67
+int nsd_journal_enumerate_available_unique(NsdJournal *j, const void **data, size_t *l);
68
#endif /* HAVE_SD_JOURNAL_RESTART_FIELDS */
69
70
int nsd_journal_add_match(NsdJournal *j, const void *data, uintptr_t size);
src/collectors/systemd-journal.plugin/systemd-journal-files.c
+12
-2
@@ -147,8 +147,8 @@ nd_journal_file_get_boot_id_annotations(NsdJournal *j __maybe_unused, struct nd_
147
148
DICTIONARY *dict = dictionary_create(DICT_OPTION_SINGLE_THREADED);
149
150
- NSD_JOURNAL_FOREACH_UNIQUE(j, data, data_length)
151
- {
150
+ nsd_journal_restart_unique(j);
151
+ while ((r = nsd_journal_enumerate_available_unique(j, &data, &data_length)) > 0) {
152
const char *key, *value;
153
size_t key_length, value_length;
154
@@ -165,6 +165,16 @@ nd_journal_file_get_boot_id_annotations(NsdJournal *j __maybe_unused, struct nd_
165
dictionary_set(dict, buf, NULL, 0);
166
}
167
168
+ if (r < 0) {
169
+ errno = -r;
170
+ internal_error(
171
+ true,
172
+ "JOURNAL: while enumerating the unique _BOOT_ID values, "
173
+ "nsd_journal_enumerate_available_unique() on file '%s' returned %d",
174
+ njf->filename,
175
+ r);
176
+ }
177
+
178
void *nothing;
179
dfe_start_read(dict, nothing)
180
{
src/collectors/systemd-journal.plugin/systemd-journal.c
+10
-3
@@ -576,12 +576,14 @@ static bool netdata_systemd_filtering_by_journal(NsdJournal *j, FACETS *facets,
576
interesting = facets_key_name_is_facet(facets, field);
577
578
if (interesting) {
579
- if (nsd_journal_query_unique(j, field) >= 0) {
579
+ int r = nsd_journal_query_unique(j, field);
580
+ if (r >= 0) {
581
bool added_this_key = false;
582
size_t added_values = 0;
583
583
- NSD_JOURNAL_FOREACH_UNIQUE(j, data, data_length)
584
- { // for each value of the key
584
+ nsd_journal_restart_unique(j);
585
+ while ((r = nsd_journal_enumerate_available_unique(j, &data, &data_length)) > 0) {
586
+ // for each value of the key
587
const char *key, *value;
588
size_t key_length, value_length;
589
@@ -614,6 +616,11 @@ static bool netdata_systemd_filtering_by_journal(NsdJournal *j, FACETS *facets,
616
added_values++;
617
filters_added++;
618
}
619
+
620
+ if (r < 0)
621
+ failures++;
622
+ } else {
623
+ failures++;
624
}
625
}
626
}
src/crates/jf/Cargo.lock
+129
-1
@@ -64,6 +64,21 @@ version = "2.10.0"
64
source = "registry+https://github.com/rust-lang/crates.io-index"
65
checksum = "812e12b5285cc515a9c72a5c1d3b6d46a19dac5acfef5265968c166106e31dd3"
66
67
+[[package]]
68
+name = "block-buffer"
69
+version = "0.10.4"
70
+source = "registry+https://github.com/rust-lang/crates.io-index"
71
+checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71"
72
+dependencies = [
73
+ "generic-array",
74
+]
75
+
76
+[[package]]
77
+name = "byteorder"
78
+version = "1.5.0"
79
+source = "registry+https://github.com/rust-lang/crates.io-index"
80
+checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b"
81
+
82
[[package]]
83
name = "cbindgen"
84
version = "0.28.0"
@@ -122,6 +137,50 @@ version = "1.0.4"
137
source = "registry+https://github.com/rust-lang/crates.io-index"
138
checksum = "b05b61dc5112cbb17e4b6cd61790d9845d13888356391624cbe7e41efeac1e75"
139
140
+[[package]]
141
+name = "cpufeatures"
142
+version = "0.2.17"
143
+source = "registry+https://github.com/rust-lang/crates.io-index"
144
+checksum = "59ed5838eebb26a2bb2e58f6d5b5316989ae9d08bab10e0e6d103e656d1b0280"
145
+dependencies = [
146
+ "libc",
147
+]
148
+
149
+[[package]]
150
+name = "crc"
151
+version = "3.3.0"
152
+source = "registry+https://github.com/rust-lang/crates.io-index"
153
+checksum = "9710d3b3739c2e349eb44fe848ad0b7c8cb1e42bd87ee49371df2f7acaf3e675"
154
+dependencies = [
155
+ "crc-catalog",
156
+]
157
+
158
+[[package]]
159
+name = "crc-catalog"
160
+version = "2.5.0"
161
+source = "registry+https://github.com/rust-lang/crates.io-index"
162
+checksum = "217698eaf96b4a3f0bc4f3662aaa55bdf913cd54d7204591faa790070c6d0853"
163
+
164
+[[package]]
165
+name = "crypto-common"
166
+version = "0.1.7"
167
+source = "registry+https://github.com/rust-lang/crates.io-index"
168
+checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a"
169
+dependencies = [
170
+ "generic-array",
171
+ "typenum",
172
+]
173
+
174
+[[package]]
175
+name = "digest"
176
+version = "0.10.7"
177
+source = "registry+https://github.com/rust-lang/crates.io-index"
178
+checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292"
179
+dependencies = [
180
+ "block-buffer",
181
+ "crypto-common",
182
+]
183
+
184
[[package]]
185
name = "equivalent"
186
version = "1.0.2"
@@ -159,6 +218,25 @@ version = "0.1.5"
218
source = "registry+https://github.com/rust-lang/crates.io-index"
219
checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2"
220
221
+[[package]]
222
+name = "fxhash"
223
+version = "0.2.1"
224
+source = "registry+https://github.com/rust-lang/crates.io-index"
225
+checksum = "c31b6d751ae2c7f11320402d34e41349dd1016f8d5d45e48c4312bc8625af50c"
226
+dependencies = [
227
+ "byteorder",
228
+]
229
+
230
+[[package]]
231
+name = "generic-array"
232
+version = "0.14.7"
233
+source = "registry+https://github.com/rust-lang/crates.io-index"
234
+checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a"
235
+dependencies = [
236
+ "typenum",
237
+ "version_check",
238
+]
239
+
240
[[package]]
241
name = "getrandom"
242
version = "0.3.4"
@@ -199,6 +277,15 @@ version = "0.16.1"
277
source = "registry+https://github.com/rust-lang/crates.io-index"
278
checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100"
279
280
+[[package]]
281
+name = "hashers"
282
+version = "1.0.1"
283
+source = "registry+https://github.com/rust-lang/crates.io-index"
284
+checksum = "b2bca93b15ea5a746f220e56587f71e73c6165eab783df9e26590069953e3c30"
285
+dependencies = [
286
+ "fxhash",
287
+]
288
+
289
[[package]]
290
name = "heck"
291
version = "0.4.1"
@@ -252,13 +339,15 @@ name = "journal_file"
339
version = "0.1.0"
340
dependencies = [
341
"error",
342
+ "hashers",
343
"hex",
344
+ "lz4_flex",
345
+ "lzma-rust2",
346
"memmap2",
347
"rand",
348
"ruzstd",
349
"siphasher",
350
"tempfile",
261
- "twox-hash",
351
"window_manager",
352
"zerocopy 0.9.0-alpha.0",
353
]
@@ -299,6 +388,22 @@ version = "0.4.29"
388
source = "registry+https://github.com/rust-lang/crates.io-index"
389
checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897"
390
391
+[[package]]
392
+name = "lz4_flex"
393
+version = "0.12.1"
394
+source = "registry+https://github.com/rust-lang/crates.io-index"
395
+checksum = "98c23545df7ecf1b16c303910a69b079e8e251d60f7dd2cc9b4177f2afaf1746"
396
+
397
+[[package]]
398
+name = "lzma-rust2"
399
+version = "0.15.7"
400
+source = "registry+https://github.com/rust-lang/crates.io-index"
401
+checksum = "1670343e58806300d87950e3401e820b519b9384281bbabfb15e3636689ffd69"
402
+dependencies = [
403
+ "crc",
404
+ "sha2",
405
+]
406
+
407
[[package]]
408
name = "memchr"
409
version = "2.8.0"
@@ -478,6 +583,17 @@ dependencies = [
583
"serde",
584
]
585
586
+[[package]]
587
+name = "sha2"
588
+version = "0.10.9"
589
+source = "registry+https://github.com/rust-lang/crates.io-index"
590
+checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283"
591
+dependencies = [
592
+ "cfg-if",
593
+ "cpufeatures",
594
+ "digest",
595
+]
596
+
597
[[package]]
598
name = "sigbus"
599
version = "0.1.0"
@@ -595,6 +711,12 @@ version = "2.1.2"
711
source = "registry+https://github.com/rust-lang/crates.io-index"
712
checksum = "9ea3136b675547379c4bd395ca6b938e5ad3c3d20fad76e7fe85f9e0d011419c"
713
714
+[[package]]
715
+name = "typenum"
716
+version = "1.20.0"
717
+source = "registry+https://github.com/rust-lang/crates.io-index"
718
+checksum = "40ce102ab67701b8526c123c1bab5cbe42d7040ccfd0f64af1a385808d2f43de"
719
+
720
[[package]]
721
name = "unicode-ident"
722
version = "1.0.23"
@@ -613,6 +735,12 @@ version = "0.2.2"
735
source = "registry+https://github.com/rust-lang/crates.io-index"
736
checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821"
737
738
+[[package]]
739
+name = "version_check"
740
+version = "0.9.5"
741
+source = "registry+https://github.com/rust-lang/crates.io-index"
742
+checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a"
743
+
744
[[package]]
745
name = "wasip2"
746
version = "1.0.2+wasi-0.2.9"
src/crates/jf/Cargo.toml
+3
-1
@@ -19,10 +19,12 @@ zerocopy = { version = "0.9.0-alpha.0", features = ["derive"] }
19
thiserror = "2"
20
rand = "0.9"
21
siphasher = "1.0"
22
-twox-hash = { version = "2.1", default-features = false, features = ["std"] }
22
+hashers = "1.0"
23
static_assertions = "1.1"
24
serde_json = "1.0"
25
ruzstd = "0.8"
26
+lz4_flex = { version = "0.12", default-features = false, features = ["std", "safe-decode", "checked-decode"] }
27
+lzma-rust2 = { version = "0.15", default-features = false, features = ["std", "xz"] }
28
libc = "0.2"
29
hex = "0.4"
30
tempfile = "3"
src/crates/jf/journal_file/Cargo.toml
+3
-1
@@ -9,8 +9,10 @@ error = { path = "../error" }
9
window_manager = { path = "../window_manager" }
10
memmap2 = { workspace = true }
11
ruzstd = { workspace = true }
12
+lz4_flex = { workspace = true }
13
+lzma-rust2 = { workspace = true }
14
siphasher = { workspace = true }
13
-twox-hash = { workspace = true }
15
+hashers = { workspace = true }
16
zerocopy = { workspace = true }
17
hex = { workspace = true }
18
rand = { workspace = true }
src/crates/jf/journal_file/src/file.rs
+186
-4
@@ -160,13 +160,38 @@ struct PayloadMatcher<'data, T> {
160
_phantom: PhantomData<T>,
161
}
162
163
-impl<'data, B: ByteSlice> PayloadMatcher<'data, DataObject<B>> {
164
- fn data_matcher(payload: &'data [u8], hash: u64) -> Self {
163
+struct DataPayloadMatcher<'data> {
164
+ payload: &'data [u8],
165
+ hash: u64,
166
+ decompressed_payload: Vec<u8>,
167
+}
168
+
169
+impl<'data> DataPayloadMatcher<'data> {
170
+ fn new(payload: &'data [u8], hash: u64) -> Self {
171
Self {
172
payload,
173
hash,
168
- _phantom: PhantomData::<DataObject<B>>,
174
+ decompressed_payload: Vec::new(),
175
+ }
176
+ }
177
+
178
+ fn payload_matches<B: ByteSlice>(&mut self, object: &DataObject<B>) -> Result<bool> {
179
+ if object.get_payload() == self.payload {
180
+ return Ok(true);
181
+ }
182
+
183
+ if object.is_compressed() {
184
+ let len = match object.decompress(&mut self.decompressed_payload) {
185
+ Ok(len) => len,
186
+ Err(JournalError::DecompressorError | JournalError::UnknownCompressionMethod) => {
187
+ return Ok(false);
188
+ }
189
+ Err(e) => return Err(e),
190
+ };
191
+ return Ok(&self.decompressed_payload[..len] == self.payload);
192
}
193
+
194
+ Ok(false)
195
}
196
}
197
@@ -180,6 +205,19 @@ impl<'data, B: ByteSlice> PayloadMatcher<'data, FieldObject<B>> {
205
}
206
}
207
208
+impl<'a, 'data> BucketVisitor<'a> for DataPayloadMatcher<'data> {
209
+ type Object = DataObject<&'a [u8]>;
210
+ type Output = NonZeroU64;
211
+
212
+ fn visit(&mut self, object: &ValueGuard<'a, Self::Object>) -> Result<Option<Self::Output>> {
213
+ if object.hash() == self.hash && self.payload_matches(object)? {
214
+ Ok(Some(object.offset()))
215
+ } else {
216
+ Ok(None)
217
+ }
218
+ }
219
+}
220
+
221
impl<'a, T> BucketVisitor<'a> for PayloadMatcher<'_, T>
222
where
223
T: JournalObject<&'a [u8]> + HashableObject,
@@ -546,7 +584,7 @@ impl<M: MemoryMap> JournalFile<M> {
584
}
585
586
pub fn find_data_offset(&self, hash: u64, payload: &[u8]) -> Result<Option<NonZeroU64>> {
549
- let visitor = PayloadMatcher::data_matcher(payload, hash);
587
+ let visitor = DataPayloadMatcher::new(payload, hash);
588
self.visit_bucket(self.data_hash_table_ref(), hash, visitor)
589
}
590
@@ -1139,3 +1177,147 @@ impl<'a, M: MemoryMap> Iterator for EntryDataIterator<'a, M> {
1177
}
1178
}
1179
}
1180
+
1181
+#[cfg(test)]
1182
+mod tests {
1183
+ use super::*;
1184
+ use crate::JournalWriter;
1185
+ use zerocopy::IntoBytes;
1186
+
1187
+ fn data_object_bytes(payload: &[u8], flags: u8) -> Vec<u8> {
1188
+ let header = DataObjectHeader {
1189
+ object_header: ObjectHeader {
1190
+ type_: ObjectType::Data as u8,
1191
+ flags,
1192
+ reserved: [0; 6],
1193
+ size: (std::mem::size_of::<DataObjectHeader>() + payload.len()) as u64,
1194
+ },
1195
+ hash: 0,
1196
+ next_hash_offset: None,
1197
+ next_field_offset: None,
1198
+ entry_offset: None,
1199
+ entry_array_offset: None,
1200
+ n_entries: None,
1201
+ };
1202
+
1203
+ let mut bytes = Vec::with_capacity(header.object_header.size as usize);
1204
+ bytes.extend_from_slice(header.as_bytes());
1205
+ bytes.extend_from_slice(payload);
1206
+ bytes
1207
+ }
1208
+
1209
+ #[test]
1210
+ fn data_payload_matcher_matches_lz4_compressed_payload() {
1211
+ let payload = b"_SYSTEMD_UNIT=netdata.service";
1212
+ let compressed = lz4_flex::block::compress(payload);
1213
+ let mut stored_payload = Vec::with_capacity(std::mem::size_of::<u64>() + compressed.len());
1214
+ stored_payload.extend_from_slice(&(payload.len() as u64).to_le_bytes());
1215
+ stored_payload.extend_from_slice(&compressed);
1216
+
1217
+ let bytes = data_object_bytes(&stored_payload, ObjectFlags::CompressedLz4 as u8);
1218
+ let object = DataObject::from_data(bytes.as_slice(), false).unwrap();
1219
+
1220
+ let mut matcher = DataPayloadMatcher::new(payload, 0);
1221
+ assert!(matcher.payload_matches(&object).unwrap());
1222
+ }
1223
+
1224
+ #[test]
1225
+ fn find_data_offset_matches_lz4_compressed_payload_in_hash_bucket() -> Result<()> {
1226
+ let payload = b"_SYSTEMD_UNIT=netdata.service";
1227
+ let temp_file = tempfile::NamedTempFile::new().map_err(JournalError::Io)?;
1228
+ let options = JournalFileOptions::new([1; 16], [2; 16], [3; 16], [4; 16]);
1229
+ let mut journal_file = JournalFile::<memmap2::MmapMut>::create(temp_file.path(), options)?;
1230
+ let data_offset = {
1231
+ let writer = JournalWriter::new(&mut journal_file)?;
1232
+ NonZeroU64::new(writer.current_file_size()).unwrap()
1233
+ };
1234
+ let hash = journal_file.hash(payload);
1235
+
1236
+ let compressed = lz4_flex::block::compress(payload);
1237
+ let mut stored_payload = Vec::with_capacity(std::mem::size_of::<u64>() + compressed.len());
1238
+ stored_payload.extend_from_slice(&(payload.len() as u64).to_le_bytes());
1239
+ stored_payload.extend_from_slice(&compressed);
1240
+
1241
+ {
1242
+ let mut data_guard =
1243
+ journal_file.data_mut(data_offset, Some(stored_payload.len() as u64))?;
1244
+ data_guard.header.hash = hash;
1245
+ data_guard.header.object_header.flags = ObjectFlags::CompressedLz4 as u8;
1246
+ data_guard.set_payload(&stored_payload);
1247
+ }
1248
+
1249
+ journal_file.data_hash_table_set_tail_offset(hash, data_offset)?;
1250
+
1251
+ assert_eq!(
1252
+ journal_file.find_data_offset(hash, payload)?,
1253
+ Some(data_offset)
1254
+ );
1255
+ assert_eq!(
1256
+ journal_file.find_data_offset(hash, b"_SYSTEMD_UNIT=sshd.service")?,
1257
+ None
1258
+ );
1259
+
1260
+ Ok(())
1261
+ }
1262
+
1263
+ #[test]
1264
+ fn find_data_offset_skips_bad_compressed_payload_in_hash_bucket() -> Result<()> {
1265
+ let payload = b"_SYSTEMD_UNIT=netdata.service";
1266
+ let temp_file = tempfile::NamedTempFile::new().map_err(JournalError::Io)?;
1267
+ let options = JournalFileOptions::new([1; 16], [2; 16], [3; 16], [4; 16]);
1268
+ let mut journal_file = JournalFile::<memmap2::MmapMut>::create(temp_file.path(), options)?;
1269
+ let bad_offset = {
1270
+ let writer = JournalWriter::new(&mut journal_file)?;
1271
+ NonZeroU64::new(writer.current_file_size()).unwrap()
1272
+ };
1273
+ let hash = journal_file.hash(payload);
1274
+
1275
+ let bad_size = {
1276
+ let mut data_guard = journal_file.data_mut(bad_offset, Some(5))?;
1277
+ data_guard.header.hash = hash;
1278
+ data_guard.header.object_header.flags = ObjectFlags::CompressedLz4 as u8;
1279
+ data_guard.set_payload(b"short");
1280
+
1281
+ data_guard.header.object_header.aligned_size()
1282
+ };
1283
+ let good_offset = NonZeroU64::new(bad_offset.get() + bad_size).unwrap();
1284
+
1285
+ let compressed = lz4_flex::block::compress(payload);
1286
+ let mut stored_payload = Vec::with_capacity(std::mem::size_of::<u64>() + compressed.len());
1287
+ stored_payload.extend_from_slice(&(payload.len() as u64).to_le_bytes());
1288
+ stored_payload.extend_from_slice(&compressed);
1289
+
1290
+ {
1291
+ let mut data_guard =
1292
+ journal_file.data_mut(good_offset, Some(stored_payload.len() as u64))?;
1293
+ data_guard.header.hash = hash;
1294
+ data_guard.header.object_header.flags = ObjectFlags::CompressedLz4 as u8;
1295
+ data_guard.set_payload(&stored_payload);
1296
+ }
1297
+
1298
+ journal_file.data_hash_table_set_tail_offset(hash, bad_offset)?;
1299
+ journal_file.data_hash_table_set_tail_offset(hash, good_offset)?;
1300
+
1301
+ assert_eq!(
1302
+ journal_file.find_data_offset(hash, payload)?,
1303
+ Some(good_offset)
1304
+ );
1305
+
1306
+ Ok(())
1307
+ }
1308
+
1309
+ #[test]
1310
+ fn data_payload_matcher_rejects_different_compressed_payload() {
1311
+ let payload = b"_SYSTEMD_UNIT=netdata.service";
1312
+ let compressed = lz4_flex::block::compress(payload);
1313
+ let mut stored_payload = Vec::with_capacity(std::mem::size_of::<u64>() + compressed.len());
1314
+ stored_payload.extend_from_slice(&(payload.len() as u64).to_le_bytes());
1315
+ stored_payload.extend_from_slice(&compressed);
1316
+
1317
+ let bytes = data_object_bytes(&stored_payload, ObjectFlags::CompressedLz4 as u8);
1318
+ let object = DataObject::from_data(bytes.as_slice(), false).unwrap();
1319
+
1320
+ let mut matcher = DataPayloadMatcher::new(b"_SYSTEMD_UNIT=sshd.service", 0);
1321
+ assert!(!matcher.payload_matches(&object).unwrap());
1322
+ }
1323
+}
src/crates/jf/journal_file/src/hash.rs
+33
-3
@@ -2,10 +2,20 @@ use siphasher::sip::SipHasher24;
2
use std::hash::Hasher;
3
4
fn jenkins_hash64(data: &[u8]) -> u64 {
5
- // FIXME: user real jenkins hasher
6
- let mut hasher = twox_hash::XxHash64::default();
5
+ use hashers::jenkins::Lookup3Hasher;
6
+
7
+ if data.is_empty() {
8
+ // systemd's jenkins_hashlittle2() starts both halves from 0xdeadbeef.
9
+ return 0xdead_beef_dead_beef;
10
+ }
11
+
12
+ let mut hasher = Lookup3Hasher::default();
13
hasher.write(data);
8
- hasher.finish()
14
+ let hash = hasher.finish();
15
+
16
+ let low = (hash & 0xFFFF_FFFF) as u32;
17
+ let high = (hash >> 32) as u32;
18
+ ((low as u64) << 32) | high as u64
19
}
20
21
fn siphash24(data: &[u8], key: &[u8; 16]) -> u64 {
@@ -29,3 +39,23 @@ pub fn journal_hash_data(data: &[u8], is_keyed_hash: bool, file_id: Option<&[u8;
39
jenkins_hash64(data)
40
}
41
}
42
+
43
+#[cfg(test)]
44
+mod tests {
45
+ use super::*;
46
+
47
+ #[test]
48
+ fn jenkins_hash64_matches_systemd_lookup3_values() {
49
+ let cases: &[(&[u8], u64)] = &[
50
+ (b"", 0xdead_beef_dead_beef),
51
+ (b"SYSLOG_IDENTIFIER=netdata", 0x45cc_d0e9_ed13_614a),
52
+ (b"_SYSTEMD_UNIT=netdata.service", 0x1013_c5df_11a9_83f0),
53
+ (b"PRIORITY=6", 0x80f0_9f19_808d_26a3),
54
+ (b"MESSAGE=Test message", 0x8ed5_3fb5_2aa5_c55d),
55
+ ];
56
+
57
+ for (payload, expected) in cases {
58
+ assert_eq!(jenkins_hash64(payload), *expected);
59
+ }
60
+ }
61
+}
src/crates/jf/journal_file/src/object.rs
+276
-6
@@ -779,6 +779,77 @@ pub struct DataObject<B: ByteSlice> {
779
pub payload: DataPayloadType<B>,
780
}
781
782
+// systemd limits journal DATA field payloads to 768 MiB; reject corrupt size prefixes before allocating.
783
+const MAX_UNCOMPRESSED_DATA_OBJECT_SIZE: usize = 768 * 1024 * 1024;
784
+const DECOMPRESSION_READ_CHUNK_SIZE: usize = 8 * 1024;
785
+const MIN_DECOMPRESSION_RESERVE_SIZE: usize = 64 * 1024;
786
+
787
+fn read_limited_to_end<R: std::io::Read>(reader: R, buf: &mut Vec<u8>) -> Result<usize> {
788
+ read_limited_to_end_with_cap(reader, buf, MAX_UNCOMPRESSED_DATA_OBJECT_SIZE)
789
+}
790
+
791
+fn read_limited_to_end_with_cap<R: std::io::Read>(
792
+ mut reader: R,
793
+ buf: &mut Vec<u8>,
794
+ max_size: usize,
795
+) -> Result<usize> {
796
+ buf.clear();
797
+ let mut chunk = [0u8; DECOMPRESSION_READ_CHUNK_SIZE];
798
+
799
+ loop {
800
+ if buf.len() == max_size {
801
+ let mut extra = [0u8; 1];
802
+ match reader.read(&mut extra) {
803
+ Ok(0) => return Ok(buf.len()),
804
+ Ok(_) => {
805
+ *buf = Vec::new();
806
+ return Err(JournalError::DecompressorError);
807
+ }
808
+ Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
809
+ Err(_) => {
810
+ *buf = Vec::new();
811
+ return Err(JournalError::DecompressorError);
812
+ }
813
+ }
814
+ }
815
+
816
+ let remaining = max_size - buf.len();
817
+ let read_len = remaining.min(chunk.len());
818
+ match reader.read(&mut chunk[..read_len]) {
819
+ Ok(0) => return Ok(buf.len()),
820
+ Ok(len) => {
821
+ let Some(required) = buf
822
+ .len()
823
+ .checked_add(len)
824
+ .filter(|required| *required <= max_size)
825
+ else {
826
+ *buf = Vec::new();
827
+ return Err(JournalError::DecompressorError);
828
+ };
829
+
830
+ if required > buf.capacity() {
831
+ let target_capacity = required
832
+ .max(buf.capacity().saturating_mul(2))
833
+ .max(MIN_DECOMPRESSION_RESERVE_SIZE)
834
+ .min(max_size);
835
+
836
+ if buf.try_reserve_exact(target_capacity - buf.len()).is_err() {
837
+ *buf = Vec::new();
838
+ return Err(JournalError::DecompressorError);
839
+ }
840
+ }
841
+
842
+ buf.extend_from_slice(&chunk[..len]);
843
+ }
844
+ Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
845
+ Err(_) => {
846
+ *buf = Vec::new();
847
+ return Err(JournalError::DecompressorError);
848
+ }
849
+ }
850
+ }
851
+}
852
+
853
impl<B: ByteSlice> std::fmt::Debug for DataObject<B> {
854
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
855
f.debug_struct("DataObject")
@@ -875,22 +946,221 @@ impl<B: ByteSlice> DataObject<B> {
946
947
if self.zstd_compressed() {
948
use ruzstd::decoding::StreamingDecoder;
878
- use ruzstd::io::Read;
949
950
let payload = self.payload_bytes();
881
- let mut decoder =
882
- StreamingDecoder::new(payload).map_err(|_| JournalError::DecompressorError)?;
951
+ let decoder = match StreamingDecoder::new(payload) {
952
+ Ok(decoder) => decoder,
953
+ Err(_) => {
954
+ *buf = Vec::new();
955
+ return Err(JournalError::DecompressorError);
956
+ }
957
+ };
958
+
959
+ read_limited_to_end(decoder, buf)
960
+ } else if self.lz4_compressed() {
961
+ let payload = self.payload_bytes();
962
+
963
+ if payload.len() < 8 {
964
+ *buf = Vec::new();
965
+ return Err(JournalError::DecompressorError);
966
+ }
967
+
968
+ let size_bytes = match payload[..8].try_into() {
969
+ Ok(size_bytes) => size_bytes,
970
+ Err(_) => {
971
+ *buf = Vec::new();
972
+ return Err(JournalError::DecompressorError);
973
+ }
974
+ };
975
+ let uncompressed_size = match usize::try_from(u64::from_le_bytes(size_bytes)) {
976
+ Ok(uncompressed_size) => uncompressed_size,
977
+ Err(_) => {
978
+ *buf = Vec::new();
979
+ return Err(JournalError::DecompressorError);
980
+ }
981
+ };
982
+ if uncompressed_size > MAX_UNCOMPRESSED_DATA_OBJECT_SIZE {
983
+ *buf = Vec::new();
984
+ return Err(JournalError::DecompressorError);
985
+ }
986
+ let compressed_data = &payload[8..];
987
988
buf.clear();
885
- decoder
886
- .read_to_end(buf)
887
- .map_err(|_| JournalError::DecompressorError)
989
+ if uncompressed_size > buf.capacity() {
990
+ if buf.try_reserve_exact(uncompressed_size).is_err() {
991
+ *buf = Vec::new();
992
+ return Err(JournalError::DecompressorError);
993
+ }
994
+ }
995
+ buf.resize(uncompressed_size, 0);
996
+
997
+ match lz4_flex::block::decompress_into(compressed_data, buf) {
998
+ Ok(len) if len == uncompressed_size => Ok(len),
999
+ Ok(_) | Err(_) => {
1000
+ *buf = Vec::new();
1001
+ Err(JournalError::DecompressorError)
1002
+ }
1003
+ }
1004
+ } else if self.xz_compressed() {
1005
+ use lzma_rust2::XzReader;
1006
+
1007
+ let payload = self.payload_bytes();
1008
+ let decoder = XzReader::new(payload, false);
1009
+
1010
+ read_limited_to_end(decoder, buf)
1011
} else {
1012
+ *buf = Vec::new();
1013
Err(JournalError::UnknownCompressionMethod)
1014
}
1015
}
1016
}
1017
1018
+#[cfg(test)]
1019
+mod tests {
1020
+ use super::*;
1021
+ use std::io::{self, Read};
1022
+
1023
+ struct FixedSizeReader {
1024
+ remaining: usize,
1025
+ }
1026
+
1027
+ impl Read for FixedSizeReader {
1028
+ fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
1029
+ let len = self.remaining.min(buf.len());
1030
+ buf[..len].fill(b'x');
1031
+ self.remaining -= len;
1032
+ Ok(len)
1033
+ }
1034
+ }
1035
+
1036
+ fn data_object_bytes(payload: &[u8], flags: u8) -> Vec<u8> {
1037
+ let header = DataObjectHeader {
1038
+ object_header: ObjectHeader {
1039
+ type_: ObjectType::Data as u8,
1040
+ flags,
1041
+ reserved: [0; 6],
1042
+ size: (std::mem::size_of::<DataObjectHeader>() + payload.len()) as u64,
1043
+ },
1044
+ hash: 0,
1045
+ next_hash_offset: None,
1046
+ next_field_offset: None,
1047
+ entry_offset: None,
1048
+ entry_array_offset: None,
1049
+ n_entries: None,
1050
+ };
1051
+
1052
+ let mut bytes = Vec::with_capacity(header.object_header.size as usize);
1053
+ bytes.extend_from_slice(header.as_bytes());
1054
+ bytes.extend_from_slice(payload);
1055
+ bytes
1056
+ }
1057
+
1058
+ #[test]
1059
+ fn lz4_decompress_clears_buffer_on_short_prefix() {
1060
+ let bytes = data_object_bytes(b"short", ObjectFlags::CompressedLz4 as u8);
1061
+ let object = DataObject::from_data(bytes.as_slice(), false).unwrap();
1062
+ let mut buf = b"stale".to_vec();
1063
+
1064
+ assert!(matches!(
1065
+ object.decompress(&mut buf),
1066
+ Err(JournalError::DecompressorError)
1067
+ ));
1068
+ assert!(buf.is_empty());
1069
+ assert_eq!(buf.capacity(), 0);
1070
+ }
1071
+
1072
+ #[test]
1073
+ fn lz4_decompress_rejects_oversized_payload_prefix() {
1074
+ let mut stored_payload = Vec::new();
1075
+ stored_payload
1076
+ .extend_from_slice(&((MAX_UNCOMPRESSED_DATA_OBJECT_SIZE as u64) + 1).to_le_bytes());
1077
+ stored_payload.extend_from_slice(b"invalid");
1078
+
1079
+ let bytes = data_object_bytes(&stored_payload, ObjectFlags::CompressedLz4 as u8);
1080
+ let object = DataObject::from_data(bytes.as_slice(), false).unwrap();
1081
+ let mut buf = b"stale".to_vec();
1082
+
1083
+ assert!(matches!(
1084
+ object.decompress(&mut buf),
1085
+ Err(JournalError::DecompressorError)
1086
+ ));
1087
+ assert!(buf.is_empty());
1088
+ assert_eq!(buf.capacity(), 0);
1089
+ }
1090
+
1091
+ #[test]
1092
+ fn lz4_decompress_clears_buffer_on_decode_error() {
1093
+ let uncompressed_size = 4usize;
1094
+ let mut stored_payload = Vec::new();
1095
+ stored_payload.extend_from_slice(&(uncompressed_size as u64).to_le_bytes());
1096
+ stored_payload.extend_from_slice(&[0x10, b'a', 1, 0]);
1097
+
1098
+ let bytes = data_object_bytes(&stored_payload, ObjectFlags::CompressedLz4 as u8);
1099
+ let object = DataObject::from_data(bytes.as_slice(), false).unwrap();
1100
+ let mut buf = b"stale".to_vec();
1101
+
1102
+ assert!(matches!(
1103
+ object.decompress(&mut buf),
1104
+ Err(JournalError::DecompressorError)
1105
+ ));
1106
+ assert!(buf.is_empty());
1107
+ assert_eq!(buf.capacity(), 0);
1108
+ }
1109
+
1110
+ #[test]
1111
+ fn lz4_decompress_rejects_size_mismatch() {
1112
+ let uncompressed_size = 4usize;
1113
+ let mut stored_payload = Vec::new();
1114
+ stored_payload.extend_from_slice(&(uncompressed_size as u64).to_le_bytes());
1115
+ stored_payload.extend_from_slice(&[0x30, b'a', b'b', b'c']);
1116
+
1117
+ let bytes = data_object_bytes(&stored_payload, ObjectFlags::CompressedLz4 as u8);
1118
+ let object = DataObject::from_data(bytes.as_slice(), false).unwrap();
1119
+ let mut buf = b"stale".to_vec();
1120
+
1121
+ assert!(matches!(
1122
+ object.decompress(&mut buf),
1123
+ Err(JournalError::DecompressorError)
1124
+ ));
1125
+ assert!(buf.is_empty());
1126
+ assert_eq!(buf.capacity(), 0);
1127
+ }
1128
+
1129
+ #[test]
1130
+ fn read_limited_to_end_errors_and_clears_when_limit_is_exceeded() {
1131
+ let mut buf = b"stale".to_vec();
1132
+
1133
+ assert!(matches!(
1134
+ read_limited_to_end_with_cap(FixedSizeReader { remaining: 5 }, &mut buf, 4),
1135
+ Err(JournalError::DecompressorError)
1136
+ ));
1137
+ assert!(buf.is_empty());
1138
+ assert_eq!(buf.capacity(), 0);
1139
+ }
1140
+
1141
+ #[test]
1142
+ fn xz_decompress_returns_payload() {
1143
+ let payload = b"_SYSTEMD_UNIT=netdata.service";
1144
+ let compressed = [
1145
+ 0xfd, 0x37, 0x7a, 0x58, 0x5a, 0x00, 0x00, 0x01, 0x69, 0x22, 0xde, 0x36, 0x04, 0xc0,
1146
+ 0x21, 0x1d, 0x21, 0x01, 0x16, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
1147
+ 0xe6, 0x6a, 0x1c, 0x77, 0x01, 0x00, 0x1c, 0x5f, 0x53, 0x59, 0x53, 0x54, 0x45, 0x4d,
1148
+ 0x44, 0x5f, 0x55, 0x4e, 0x49, 0x54, 0x3d, 0x6e, 0x65, 0x74, 0x64, 0x61, 0x74, 0x61,
1149
+ 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x00, 0x00, 0x00, 0x00, 0x11, 0x15,
1150
+ 0x71, 0xd5, 0x00, 0x01, 0x39, 0x1d, 0x48, 0x54, 0x04, 0x4d, 0x90, 0x42, 0x99, 0x0d,
1151
+ 0x01, 0x00, 0x00, 0x00, 0x00, 0x01, 0x59, 0x5a,
1152
+ ];
1153
+
1154
+ let bytes = data_object_bytes(&compressed, ObjectFlags::CompressedXz as u8);
1155
+ let object = DataObject::from_data(bytes.as_slice(), false).unwrap();
1156
+ let mut buf = Vec::new();
1157
+
1158
+ let len = object.decompress(&mut buf).unwrap();
1159
+
1160
+ assert_eq!(&buf[..len], payload);
1161
+ }
1162
+}
1163
+
1164
// SHA-256 HMAC is 32 bytes (256 bits)
1165
pub const TAG_LENGTH: usize = 256 / 8;
1166
src/crates/jf/journal_file/src/writer.rs
+6
-2
@@ -633,8 +633,12 @@ mod tests {
633
filtered_entries += 1;
634
}
635
636
- // Should find 2 entries with _SYSTEMD_UNIT=test.service (entries 0 and 2)
637
- assert_eq!(filtered_entries, 2, "Expected 2 filtered entries");
636
+ // Should find 2 entries with _SYSTEMD_UNIT=test.service per iteration (entries 0 and 2)
637
+ assert_eq!(
638
+ filtered_entries,
639
+ 2 * iterations,
640
+ "Expected 2 entries with _SYSTEMD_UNIT=test.service per iteration"
641
+ );
642
}
643
644
println!("✅ All tests passed!");
src/crates/jf/journal_reader_ffi/src/lib.rs
+15
-8
@@ -291,11 +291,7 @@ unsafe extern "C" fn rsd_journal_enumerate_available_data(
291
*data = journal.decompressed_payload.as_ptr() as *const c_void;
292
1
293
}
294
- Err(error::JournalError::UnknownCompressionMethod) => {
295
- eprintln!("unknown compression method");
296
- -1
297
- }
298
- Err(_) => -1,
294
+ Err(e) => e.to_error_code(),
295
};
296
} else {
297
let payload = data_guard.payload_bytes();
@@ -382,9 +378,20 @@ unsafe extern "C" fn rsd_journal_enumerate_available_unique(
378
379
match journal.reader.field_data_enumerate(&journal.journal_file) {
380
Ok(Some(data_guard)) => {
385
- let payload = data_guard.payload_bytes();
386
- *data = payload.as_ptr() as *const c_void;
387
- *l = payload.len();
381
+ if data_guard.is_compressed() {
382
+ return match data_guard.decompress(&mut journal.decompressed_payload) {
383
+ Ok(n) => {
384
+ *l = n;
385
+ *data = journal.decompressed_payload.as_ptr() as *const c_void;
386
+ 1
387
+ }
388
+ Err(e) => e.to_error_code(),
389
+ };
390
+ } else {
391
+ let payload = data_guard.payload_bytes();
392
+ *data = payload.as_ptr() as *const c_void;
393
+ *l = payload.len();
394
+ }
395
396
1
397
}