diff --git a/.agents/sow/SOW-status.md b/.agents/sow/SOW-status.md index f62c9314..698be162 100644 --- a/.agents/sow/SOW-status.md +++ b/.agents/sow/SOW-status.md @@ -1,6 +1,6 @@ # SOW Status -Last updated: 2026-11-08 +Last updated: 2026-11-09 ## Backlog Order @@ -82,6 +82,10 @@ order; the user may reorder it. ## Recently Closed Or Completed +- SOW-0178 - Undecompressible DATA Costs One Field, Not The Entry: completed + 2026-11-09. Readers drop only the damaged field, as systemd does for zstd and + lz4; for corrupt xz the SDK intentionally keeps the entry where systemd fails + the read. - SOW-0176 - Single Ownership Follow-ups: completed 2026-11-08; follow-up SOW-0177. - SOW-0174 - Rust Registry Unwatch Leaves Files When The Directory Key diff --git a/.agents/sow/done/SOW-0178-20261109-undecompressible-data-costs-one-field.md b/.agents/sow/done/SOW-0178-20261109-undecompressible-data-costs-one-field.md new file mode 100644 index 00000000..7b64b663 --- /dev/null +++ b/.agents/sow/done/SOW-0178-20261109-undecompressible-data-costs-one-field.md @@ -0,0 +1,307 @@ +# SOW-0178 - Undecompressible DATA Costs One Field, Not The Entry + +## Status + +Status: completed + +`completed` is the successful terminal status. `done` is a directory name, not a status value. Do not use `Status: done` or `Status: complete`. + +Sub-state: completed 2026-11-09 (user decisions 1B and xz = B). Branch `sow-0178/undecompressible-data-field`, +stacked on SOW-0176. + +## Requirements + +### Purpose + +A damaged compressed DATA object should cost the field it holds, as it does in +systemd, instead of failing the whole entry. + +### User Request + +SOW-0176 PR review, decision 1: "B: skip the field, as systemd does", as its +own SOW so the reader-hardening contract change is reviewed separately. + +### Assistant Understanding + +Facts: + +- systemd skips an entry item whose payload read returns `-EADDRNOTAVAIL` or + `-EBADMSG` (systemd/systemd @ c0a5a2516d28601fb3afc1a77d7b42fcfe38fced + `src/libsystemd/sd-journal/sd-journal.c:2852,2896`). Corrupt zstd and lz4 + streams return `-EBADMSG` (`src/basic/compress.c` lz4 and + `zstd_ret_to_errno`). Corrupt xz returns `-ENOMEM` + (`decompress_blob_xz`), which is not skipped. +- Baseline (before this SOW) stock v262 differential + (`.local/sow0178/codec_diff.py`, 5-entry files whose one compressed DATA + object is shared by every entry); post-change results are under Validation: + - zstd corrupt (bit flip, zeroed tail): stock rc 0, 5 rows, field dropped + (75 to 70 fields). Go and Rust: rc 1, no rows. + - lz4 bad stored size (+1, -1): stock rc 0, 5 rows, field dropped. Go and + Rust: rc 1, no rows. + - xz corrupt: stock rc 1, `Failed to read journal: Cannot allocate memory`, + no rows. Go and Rust: rc 1, no rows. + - Baseline reader-hardening zstd fixtures (bad later checksum, truncated + later frame, trailing garbage, oversized 768 MiB + 1): stock keeps all 4 + rows and drops only the damaged MESSAGE field, while the matrix then + expected an SDK entry error. +- Unique-value enumeration returns decompression errors in systemd + (`sd-journal.c` `sd_journal_enumerate_unique`), so it is out of scope. +- SDK sites: Go `recoverableEntryDataError` (`go/journal/reader_entry.go`) and + `decompressDataPayload` (`go/journal/verify_graph.go:1166`); Rust + `recoverable_entry_data_error` (`rust/src/journal/src/reader_helpers.rs:245`) + and `JournalError::DecompressorError`. +- `tests/interoperability/run_reader_hardening.py` streams cases expect an + entry error for the corrupt and oversized cases. + +Unknowns: + +- None beyond the open decision. + +### Acceptance Criteria + +- In both languages, current-entry reads (`GetEntry`/`get_entry`, payload + visitors, entry enumeration, facade data getters, Netdata row expansion) + drop a field whose compressed DATA fails to decompress, per the decision. +- Strict verification still fails such files. +- The hardening streams cases expect stock's rows, and the codec differential + matches stock exit status and row and field counts. + +## Analysis + +See Facts. The change is a classification of decompression errors in one +predicate per language, plus the consequences in the hardening matrix, tests +and docs (`docs/Reader-APIs.md`, `product-scope.md`). `docs/Rust-API.md` +needs no change: it only says unique enumeration fails on undecompressible +DATA, which stays true. + +## Pre-Implementation Gate + +Status: ready + +Problem / root-cause model: + +- Decompression errors are not in the recoverable set in either language. + +Evidence reviewed: + +- See Facts. + +Affected contracts and surfaces: + +- Go and Rust reader current-entry APIs, facades, Netdata row expansion, both + journalctl rewrites through them. + +Existing patterns to reuse: + +- The SOW-0176 recoverable-error predicates and damage tests. + +Risk and blast radius: + +- A row can silently lose a field; the field-level drop is what systemd and the + Netdata plugin already show. Verification stays strict. + +Sensitive data handling plan: + +- Synthetic fixtures only. + +Implementation plan: + +1. Classify decompression failures per the decision in both languages. +2. Paired damage tests per codec; update hardening expectations; docs and spec. + +Validation plan: + +- Unit tests, `run_reader_hardening.py`, `codec_diff.py`, query and verify + matrices. + +Artifact impact plan: + +- AGENTS.md: none expected. +- Runtime project skills: none expected. +- Specs: `product-scope.md` recovery text. +- End-user/operator docs: `docs/Reader-APIs.md`, `docs/Rust-API.md`. +- End-user/operator skills: none. +- SOW lifecycle: pending until the decision. + +Open-source reference evidence: + +- Cited in Facts. + +Open decisions: + +- None; xz decided 2026-11-09 (B). + +## Implications And Decisions + +- 2026-11-09, decision 1B (SOW-0176 review): skip the field, as systemd does. +- 2026-11-09, xz = B (user): systemd fails the whole read on corrupt xz + (`-ENOMEM`) while skipping corrupt zstd and lz4; the SDK skips the field for + every codec. The xz difference is recorded in docs, spec and code comments. + +## Plan + +See the implementation plan above. + +## Implementation And Review Plan + +Implementation: + +- The project manager implements both languages in one chunk with tests. + +Failure handling: + +- A failing test, matrix or audit is recorded in the Execution Log or + Validation with its cause, fixed or rerun, and blocks `completed` until it + passes; a flaky failure is rerun and both results are recorded. + +Reviewers: + +- Whole-SOW self-review after local validation; tests are the gate. External + reviewers are not used. + +Repository boundary block for every external-agent prompt: + +```text +CRITICAL REPOSITORY BOUNDARY: +- Do not make changes outside this repository for any reason. +- Repository path: current repository root. +- You may inspect external references read-only when the task requires it. +- Write, edit, delete, move, reset, checkout, install, generate, cache, or format nothing outside this repository. +- The only write exception outside the repository is /tmp. +- Prefer .local/ inside this repository for scratch work, generated temporary files, cloned references, logs, and working notes. +``` + +## Execution Log + +### 2026-11-09 + +- Created with the stock differential above. +- Implemented: Go wraps every decompression failure in `errDataDecompression` + (`decompressDataPayload`), which `recoverableEntryDataError` accepts. Rust + moves the predicate to `JournalError::is_recoverable_entry_data` in + `journal-core` (one owner for the SDK and `journal-engine`) and adds + `DecompressorError`. Explorer facet classification (Go + `classifyDataForAccumulator`, Rust `classify_data_for_accumulator`) and the + Rust `journal-engine` row projection skip the field too. Go Netdata rows go + through `GetEntry`/`CollectEntryPayloads`, already covered. +- `TestNativeFilterDoesNotDecodeUnrelatedCompressedData` proved a payload read + decodes the damaged DATA by expecting `GetEntry` to fail; it now checks that + MESSAGE is dropped, which proves the same thing. +- Duplication gate: the new Rust `impl` first lengthened a baselined block shared + with the legacy `jf` copy; placing it at the end of `error.rs` fixed it. +- PR review round (8 bot findings, all accepted): spec sentence scoped to tail + recovery; the stock oracle check in `run_reader_hardening.py` now parses + `-o export` and compares realtime and payloads per row instead of counts; + both explorer tests assert MESSAGE is absent from the row and reported only + as unset; Go keeps the decompressor error in the chain (`%w: %w`, Go 1.20+); + SOW baseline facts labeled, docs list corrected, failure handling recorded, + status entry states the outcome. Reruns: Go and Rust damage tests pass; + `run_reader_hardening.py` PASS, 552 observations, adapters rebuilt. + +## Validation + +Acceptance criteria evidence: + +- Current-entry reads, visitors, enumeration, facade getters, Explorer facets + and Rust row projection drop the damaged field: paired tests + `TestDamagedCompressedDataCostsTheFieldNotTheEntry` / + `damaged_compressed_data_costs_the_field` (zstd, xz, lz4) and + `TestExplorerSkipsDamagedCompressedData` / + `explorer_skips_damaged_compressed_data`. Both explorer tests fail with the + explorer change disabled. +- Strict verification still fails: asserted in the same tests and in the + hardening matrix (`verify` probe and stock verify). +- Stock parity (`.local/sow0178/codec_diff.py`): zstd bit flip and zeroed tail, + lz4 stored size +1 and -1: stock, Go and Rust all rc 0, 5 rows, 70 of 75 + fields. xz: Go and Rust rc 0, 5 rows, 70 fields; stock rc 1 (decided + difference). + +Tests or equivalent validation (2026-11-09, stock = systemd 262 oracle): + +- `go test ./journal ./cmd/...` PASS; `cargo test --workspace` PASS; + `cargo fmt --check` clean; no new clippy warnings in changed code. +- `run_reader_hardening.py` (both adapters, v262 `--journalctl`, new `.local` + `--out`, all groups): PASS, 552 observations; the streams cases now expect + stock's rows and check them against stock `-o export`. +- `run_compression_matrix.py` 24/24, `run_verify_matrix.py` PASS, + `run_mixed_directory_matrix.py` 42/42, `run_live_matrix.py` 18/18, + `verify_examples.py` and `check_wiki_docs.py` PASS, `check_duplication.py` + PASS. +- `run_journalctl_query_matrix.py` with follow: first run had one failure, + `follow-grep-boot-events` with the stock reader ("go live fixture writer did + not become ready", fixture start-up timing); the rerun passed with 0 failures. + +Real-use evidence: + +- Both built CLIs against stock v262 on the corrupted codec files above. + +Reviewer findings: + +- Project-manager review of the whole change; no external reviewers. Unique + enumeration (`EnumerateUniquePayload`, `enumerate_unique_payload`, Explorer + indexed values) still returns the error, as systemd's + `sd_journal_enumerate_unique` does. `UnknownCompressionMethod` stays an + error (systemd: `-EPROTONOSUPPORT`, not skipped by `enumerate_data`). + +Same-failure scan: + +- Every caller of `decompressDataPayload` and `.decompress(` was checked: + writer dedup and verification keep failing (the wrapped error still fails + them); unique and boot-listing paths keep the error by design. + +Sensitive data gate: + +- Synthetic fixtures and public systemd references only. + +Artifact maintenance gate: + +- AGENTS.md: no change. +- Runtime project skills: no change; the hardening recipe is unchanged. +- Specs: `product-scope.md` (field-level DATA damage). +- End-user/operator docs: `docs/Reader-APIs.md`. +- End-user/operator skills: none exist. +- SOW lifecycle: `Status: completed`, moved to `done/`. +- SOW-status.md: updated. + +Specs update: + +- `product-scope.md`. + +Project skills update: + +- None needed. + +End-user/operator docs update: + +- `docs/Reader-APIs.md`. + +End-user/operator skills update: + +- None exist. + +Lessons: + +- See `## Lessons Extracted`. + +Follow-up mapping: + +- None. + +## Outcome + +A damaged compressed DATA object now costs one field in both SDKs and CLIs, +matching stock v262 for zstd and lz4; xz deliberately drops the field where +stock fails the read. + +## Lessons Extracted + +- "As systemd does" must be checked per code path: systemd's own errno choice + (xz `-ENOMEM`) decided its behavior, not the kind of damage. + +## Followup + +None. + +## Regression Log + +None. diff --git a/.agents/sow/specs/product-scope.md b/.agents/sow/specs/product-scope.md index 0b21a34b..95f99f25 100644 --- a/.agents/sow/specs/product-scope.md +++ b/.agents/sow/specs/product-scope.md @@ -804,10 +804,18 @@ Rust applies this policy in JournalReader without changing the public JournalCursor layout; the low-level cursor handles positioning and initial error recovery, not automatic EOF refresh. +A damaged DATA object behind one entry item (bad offset, type or size, or a +compressed payload that cannot be decoded) costs that field in current-entry +reads, Explorer and Netdata facets and journalctl output, in both languages. +This matches systemd for zstd and lz4; systemd fails the whole read on corrupt +xz (`-ENOMEM`), an accepted difference (SOW-0178). Unique enumeration and strict +verification still fail. + Recovery retains native index/header access and does not expand DATA rows to -establish the prefix. Interior holes followed by readable entries, wrong types, -cycles, malformed nonzero sizes, decompression failures and general I/O errors -remain errors. Unknown secondary posting tails that could contain surviving +establish the prefix. Tail recovery does not reclassify interior holes followed +by readable entries, wrong types, cycles, malformed nonzero sizes, decompression +failures or general I/O errors as an interrupted append: outside the field-local +DATA cases above, they remain errors. Unknown secondary posting tails that could contain surviving matches remain errors. Strict verification, excluded-writer IndexedSnapshot capture, append-open and retention preconditions remain separate and strict. diff --git a/docs/Reader-APIs.md b/docs/Reader-APIs.md index f687041d..37af126e 100644 --- a/docs/Reader-APIs.md +++ b/docs/Reader-APIs.md @@ -108,11 +108,13 @@ Detection is lazy where the native access path permits it. A zero/`None` status means no loss has been observed; it does not certify integrity. A catalog query does not validate unrelated entry arrays. -A single unreadable DATA object, or one without a field separator, costs that -field only when reading the current entry: `GetEntry`/`get_entry` still return -the entry and the payload visitor and enumeration calls continue, and a payload -without `=` stays in the raw payload list. A DATA object that cannot be decompressed still fails the -entry. +A single unreadable DATA object, one that cannot be decompressed, or one without +a field separator costs that field only when reading the current entry: +`GetEntry`/`get_entry` still return the entry, the payload visitor and +enumeration calls continue, Explorer and Netdata facets skip the field, and a +payload without `=` stays in the raw payload list. Unique-value enumeration and +`--verify` still report the damage. Corrupt zstd and lz4 data match systemd; +for corrupt xz data systemd fails the whole read, while the SDK drops the field. Recovery is read-only: the SDK does not reconstruct or overwrite damaged source journals. Journald abandons a corrupt file and writes to a new one. Normal live diff --git a/go/journal/explorer.go b/go/journal/explorer.go index fdf6965a..87cf55a6 100644 --- a/go/journal/explorer.go +++ b/go/journal/explorer.go @@ -2140,11 +2140,17 @@ func (r *Reader) classifyDataForAccumulator(dataOffset uint64, acc *explorerAccu stats.DataCacheMisses++ stats.DataPayloadsLoaded++ header, err := r.readDataHeaderAt(dataOffset) - if err != nil { - return offsetClass{}, err + var payload []byte + if err == nil { + payload, err = r.readDataPayloadWithHeader(dataOffset, header) } - payload, err := r.readDataPayloadWithHeader(dataOffset, header) if err != nil { + // A bad DATA object costs its field, as in the entry readers. + if recoverableEntryDataError(err) { + class := offsetClass{kind: offsetClassIrrelevant} + acc.offsetCache.insert(dataOffset, class) + return class, nil + } return offsetClass{}, err } if header.object.flag&(objectCompressedXZ|objectCompressedLZ4|objectCompressedZSTD) != 0 { diff --git a/go/journal/reader_decompress_damage_test.go b/go/journal/reader_decompress_damage_test.go new file mode 100644 index 00000000..c2e767fb --- /dev/null +++ b/go/journal/reader_decompress_damage_test.go @@ -0,0 +1,117 @@ +package journal + +import ( + "bytes" + "encoding/binary" + "os" + "path/filepath" + "testing" +) + +// Paired with Rust tests/entry_damage.rs damaged_compressed_data_costs_the_field. +func TestDamagedCompressedDataCostsTheFieldNotTheEntry(t *testing.T) { + for _, c := range []struct { + name string + compression int + }{{"zstd", CompressionZSTD}, {"xz", CompressionXZ}, {"lz4", CompressionLZ4}} { + t.Run(c.name, func(t *testing.T) { + path := filepath.Join(t.TempDir(), "damage.journal") + opts := testOptions() + opts.Compression = c.compression + opts.CompressThresholdBytes = 16 + w, err := Create(path, opts) + if err != nil { + t.Fatal(err) + } + appendWriterFields(t, w, "entry", []Field{ + StringField("MESSAGE", "kept"), + {Name: "BROKEN", Value: bytes.Repeat([]byte("z"), 512)}, + }, EntryOptions{RealtimeUsec: 1_700_000_000_000_001, MonotonicUsec: 1}) + closeWriterForTest(t, w, "entry") + corruptCompressedData(t, path, c.compression == CompressionLZ4) + + r, err := OpenFile(path) + if err != nil { + t.Fatal(err) + } + defer r.Close() + if ok, err := r.Step(); err != nil || !ok { + t.Fatalf("step = %v, %v", ok, err) + } + entry, err := r.GetEntry() + if err != nil { + t.Fatal(err) + } + if string(entry.Fields["MESSAGE"]) != "kept" || len(entry.Payloads) != 1 { + t.Fatalf("fields = %v, payloads = %d", entry.Fields, len(entry.Payloads)) + } + payloads, err := r.CollectEntryPayloads() + if err != nil || len(payloads) != 1 { + t.Fatalf("collected = %d, %v", len(payloads), err) + } + if err := VerifyFile(path); err == nil { + t.Fatal("verification accepted a damaged compressed DATA object") + } + }) + } +} + +// corruptCompressedData breaks the only compressed DATA object: lz4 by its +// stored size, the others by their trailing checksum or footer. +func corruptCompressedData(t *testing.T, path string, lz4 bool) { + t.Helper() + data := readFileForTest(t, path) + offset := binary.LittleEndian.Uint64(data[88:]) + tail := binary.LittleEndian.Uint64(data[136:]) + for offset != 0 && offset <= tail { + size := binary.LittleEndian.Uint64(data[offset+8:]) + if data[offset] == objectTypeData && data[offset+1]&objectCompressedMask != 0 { + start, end := offset+objectHeaderToPayload, offset+size + if lz4 { + binary.LittleEndian.PutUint64(data[start:], binary.LittleEndian.Uint64(data[start:])+1) + } else { + copy(data[end-8:end], make([]byte, 8)) + } + if err := os.WriteFile(path, data, 0o644); err != nil { + t.Fatal(err) + } + return + } + offset += (size + 7) &^ 7 + } + t.Fatal("no compressed DATA object") +} + +// Paired with Rust explorer_skips_damaged_compressed_data. +func TestExplorerSkipsDamagedCompressedData(t *testing.T) { + large := append([]byte("MESSAGE="), bytes.Repeat([]byte("z"), 512)...) + path := createExplorerRawJournalWithOptions(t, []explorerTestEntry{ + {realtime: 1_000, payloads: [][]byte{[]byte("PRIORITY=3"), large}}, + }, func(opts *Options) { + opts.Compression = CompressionZSTD + opts.CompressThresholdBytes = 32 + }) + corruptCompressedData(t, path, false) + reader := mustOpenReaderFile(t, path) + defer reader.Close() + + query := DefaultExplorerQuery().WithFacet([]byte("PRIORITY")).WithFacet([]byte("MESSAGE")) + query.UseSourceRealtime = false + query.Limit = 10 + result, err := reader.Explore(query) + if err != nil { + t.Fatalf("Explore() error = %v", err) + } + assertExplorerFacetCount(t, result, "PRIORITY", "3", 1) + if values := result.Facets["MESSAGE"]; len(values) != 1 || values[string(explorerUnsetValue)] != 1 { + t.Fatalf("MESSAGE facet = %#v, want only unset=1", values) + } + if len(result.Rows) != 1 { + t.Fatalf("rows = %d, want 1", len(result.Rows)) + } + for _, payload := range result.Rows[0].Payloads { + if bytes.HasPrefix(payload, []byte("MESSAGE=")) { + t.Fatalf("row kept damaged MESSAGE payload %q", payload[:min(len(payload), 32)]) + } + } +} diff --git a/go/journal/reader_entry.go b/go/journal/reader_entry.go index e8c6fe2e..4f71e7e9 100644 --- a/go/journal/reader_entry.go +++ b/go/journal/reader_entry.go @@ -218,10 +218,11 @@ func formatCursorFromDirectoryKey(key directoryEntryKey) string { } // recoverableEntryDataError reports a bad DATA object that costs one field, not -// the entry; sd_journal_enumerate_data skips those too (sd-journal.c). +// the entry; sd_journal_enumerate_data skips those too (sd-journal.c). Corrupt +// xz is included although systemd reports it as -ENOMEM (SOW-0178). func recoverableEntryDataError(err error) bool { var bounds readerBoundsError - return errors.Is(err, errCorruptObject) || errors.As(err, &bounds) + return errors.Is(err, errCorruptObject) || errors.Is(err, errDataDecompression) || errors.As(err, &bounds) } func (r *Reader) readDataPayloadTemp(offset uint64) ([]byte, error) { diff --git a/go/journal/reader_filter_native_test.go b/go/journal/reader_filter_native_test.go index b5fffbf8..8db73c73 100644 --- a/go/journal/reader_filter_native_test.go +++ b/go/journal/reader_filter_native_test.go @@ -537,7 +537,12 @@ func TestNativeFilterDoesNotDecodeUnrelatedCompressedData(t *testing.T) { if got, err := candidate.GetRealtimeUsec(); err != nil || got != 1000 { t.Fatalf("cursor time=%d/%v", got, err) } - if _, err := candidate.GetEntry(); err == nil { + // Reading the payload decodes it; the malformed field is then dropped. + entry, err := candidate.GetEntry() + if err != nil { + t.Fatal(err) + } + if _, ok := entry.Fields["MESSAGE"]; ok { t.Fatal("payload control did not detect malformed compression") } } diff --git a/go/journal/verify_graph.go b/go/journal/verify_graph.go index 67128b03..e1c55bfb 100644 --- a/go/journal/verify_graph.go +++ b/go/journal/verify_graph.go @@ -2,6 +2,7 @@ package journal import ( "bytes" + "errors" "fmt" "math/bits" ) @@ -1163,17 +1164,26 @@ func (v *graphVerifier) hash(payload []byte) uint64 { return jenkinsHash64(payload) } +// errDataDecompression marks a compressed DATA object that cannot be decoded. +var errDataDecompression = errors.New("cannot decompress DATA object") + func decompressDataPayload(flags uint8, payload []byte) ([]byte, error) { + var out []byte + var err error switch { case flags&objectCompressedZSTD != 0: - return zstdDecompress(payload) + out, err = zstdDecompress(payload) case flags&objectCompressedXZ != 0: - return xzDecompress(payload) + out, err = xzDecompress(payload) case flags&objectCompressedLZ4 != 0: - return lz4Decompress(payload) + out, err = lz4Decompress(payload) default: return payload, nil } + if err != nil { + return nil, fmt.Errorf("%w: %w", errDataDecompression, err) + } + return out, nil } // Build object-specific diagnostic labels only when an offset is invalid. diff --git a/rust/src/crates/journal-core/src/error.rs b/rust/src/crates/journal-core/src/error.rs index e23701e6..ff19328f 100644 --- a/rust/src/crates/journal-core/src/error.rs +++ b/rust/src/crates/journal-core/src/error.rs @@ -130,3 +130,19 @@ impl From> for JournalEr } pub type Result = std::result::Result; + +impl JournalError { + /// A bad DATA object behind one entry item: readers drop that field and + /// keep the entry, as sd_journal_enumerate_data does for -EBADMSG. Corrupt + /// xz is included although systemd reports it as -ENOMEM (SOW-0178). + pub fn is_recoverable_entry_data(&self) -> bool { + matches!( + self, + JournalError::InvalidOffset + | JournalError::InvalidObjectSize(_) + | JournalError::InvalidObjectType + | JournalError::ObjectExceedsFileBounds + | JournalError::DecompressorError + ) + } +} diff --git a/rust/src/crates/journal-engine/src/logs/query.rs b/rust/src/crates/journal-engine/src/logs/query.rs index 4374a702..985e3300 100644 --- a/rust/src/crates/journal-engine/src/logs/query.rs +++ b/rust/src/crates/journal-engine/src/logs/query.rs @@ -612,7 +612,13 @@ fn read_entry_fields( row.restart_data()?; let result = (|| { - while let Some((_, payload)) = row.read_next_payload_with_offset(journal_file)? { + loop { + let payload = match row.read_next_payload_with_offset(journal_file) { + Ok(Some((_, payload))) => payload, + Ok(None) => break, + Err(err) if err.is_recoverable_entry_data() => continue, + Err(err) => return Err(err.into()), + }; let payload = row.payload_slice(payload); if let Some(pair) = read_projected_pair(payload, output_fields) { fields.push(pair); diff --git a/rust/src/journal/src/explorer.rs b/rust/src/journal/src/explorer.rs index ae2fc3f9..a32b0a02 100644 --- a/rust/src/journal/src/explorer.rs +++ b/rust/src/journal/src/explorer.rs @@ -2727,7 +2727,17 @@ fn classify_data_for_accumulator( return Ok(class); } - let payload = payload_for_classification(file, row, data_offset, stats)?; + let payload = match payload_for_classification(file, row, data_offset, stats) { + Ok(payload) => payload, + // A bad DATA object costs its field, as in the entry readers. + Err(SdkError::Journal(err)) if err.is_recoverable_entry_data() => { + accumulator + .offset_cache + .insert(data_offset, OffsetClass::Irrelevant); + return Ok(OffsetClass::Irrelevant); + } + Err(err) => return Err(err), + }; let Some((field, value)) = split_payload_bytes(payload) else { let class = classify_unstructured_payload(payload, needs_fts, query, stats); accumulator.offset_cache.insert(data_offset, class); diff --git a/rust/src/journal/src/reader_helpers.rs b/rust/src/journal/src/reader_helpers.rs index 309fbfed..c699bcd3 100644 --- a/rust/src/journal/src/reader_helpers.rs +++ b/rust/src/journal/src/reader_helpers.rs @@ -243,13 +243,7 @@ pub(super) fn verify_entry_at_strict( } pub(super) fn recoverable_entry_data_error(err: &JournalError) -> bool { - matches!( - err, - JournalError::InvalidOffset - | JournalError::InvalidObjectSize(_) - | JournalError::InvalidObjectType - | JournalError::ObjectExceedsFileBounds - ) + err.is_recoverable_entry_data() } pub(super) fn is_zst_file(path: &Path) -> bool { diff --git a/rust/src/journal/src/tests/entry_damage.rs b/rust/src/journal/src/tests/entry_damage.rs index 563235c5..8aa7e361 100644 --- a/rust/src/journal/src/tests/entry_damage.rs +++ b/rust/src/journal/src/tests/entry_damage.rs @@ -139,3 +139,120 @@ fn missing_main_entry_array_hides_every_entry() { assert!(!reader.next().unwrap(), "filtered={filtered}"); } } + +// Paired with Go TestDamagedCompressedDataCostsTheFieldNotTheEntry. +#[test] +fn damaged_compressed_data_costs_the_field() { + use journal_core::file::Compression; + + for compression in [Compression::Zstd, Compression::Xz, Compression::Lz4] { + let dir = tempfile::tempdir().unwrap(); + let path = journal_path(dir.path()); + let broken = [b"BROKEN=".as_slice(), &[b'z'; 512]].concat(); + write_damaged_compressed_entry(&path, compression, &[b"MESSAGE=kept", &broken]); + + let mut reader = DirectoryReader::open_files([&path]).unwrap(); + assert!(reader.next().unwrap(), "{compression:?}"); + let entry = reader + .get_entry() + .unwrap_or_else(|e| panic!("{compression:?}: {e:?}")); + assert_eq!(entry.fields["MESSAGE"], b"kept", "{compression:?}"); + assert_eq!(entry.payloads.len(), 1, "{compression:?}"); + let mut collected = Vec::new(); + reader.collect_entry_payloads(&mut collected).unwrap(); + assert_eq!(collected.len(), 1, "{compression:?}"); + assert!( + crate::verify_file(&path).is_err(), + "{compression:?}: verification accepted damaged DATA" + ); + } +} + +// Paired with Go TestExplorerSkipsDamagedCompressedData. +#[test] +fn explorer_skips_damaged_compressed_data() { + use journal_core::file::Compression; + + let dir = tempfile::tempdir().unwrap(); + let path = journal_path(dir.path()); + let large = [b"MESSAGE=".as_slice(), &[b'z'; 512]].concat(); + write_damaged_compressed_entry(&path, Compression::Zstd, &[b"PRIORITY=3", &large]); + + let mut reader = FileReader::open(&path).expect("open reader"); + let query = ExplorerQuery { + facets: vec![b"PRIORITY".to_vec(), b"MESSAGE".to_vec()], + use_source_realtime: false, + limit: 10, + ..ExplorerQuery::default() + }; + let result = reader.explore(&query).expect("explore"); + let priority = result + .facets + .get(b"PRIORITY".as_slice()) + .expect("priority facet"); + assert_eq!(priority.get(b"3".as_slice()), Some(&1)); + let message = result + .facets + .get(b"MESSAGE".as_slice()) + .expect("message facet"); + assert_eq!(message.len(), 1, "MESSAGE facet must hold only unset"); + assert_eq!(message.get(b"-".as_slice()), Some(&1)); + assert_eq!(result.rows.len(), 1); + assert!( + !result.rows[0] + .payloads + .iter() + .any(|payload| payload.starts_with(b"MESSAGE=")), + "row kept the damaged MESSAGE payload" + ); +} + +/// Writes one entry whose large field is compressed, then damages that DATA. +fn write_damaged_compressed_entry( + path: &Path, + compression: journal_core::file::Compression, + payloads: &[&[u8]], +) { + std::fs::create_dir_all(path.parent().unwrap()).unwrap(); + let repo_file = RepoFile::from_path(path).expect("journal path"); + let mut file = JournalFile::::create( + &repo_file, + JournalFileOptions::new(test_uuid(1), test_uuid(2), test_uuid(3)) + .with_compression(compression) + .with_compress_threshold(16), + ) + .expect("create journal"); + let mut writer = JournalWriter::new(&mut file, 1, test_uuid(4)).expect("writer"); + writer + .add_entry(&mut file, payloads, 1_700_000_000_000_000, 1) + .expect("write entry"); + file.sync().expect("sync"); + drop(file); + let lz4 = matches!(compression, journal_core::file::Compression::Lz4); + corrupt_compressed_data(path, lz4); +} + +/// Breaks the only compressed DATA object: lz4 by its stored size, the others +/// by their trailing checksum or footer. +fn corrupt_compressed_data(path: &Path, lz4: bool) { + let mut bytes = std::fs::read(path).unwrap(); + let u64_at = |b: &[u8], at: usize| u64::from_le_bytes(b[at..at + 8].try_into().unwrap()); + let mut offset = u64_at(&bytes, 88) as usize; + let tail = u64_at(&bytes, 136) as usize; + while offset != 0 && offset <= tail { + let size = u64_at(&bytes, offset + 8) as usize; + if bytes[offset] == 1 && bytes[offset + 1] & 7 != 0 { + let (start, end) = (offset + OBJECT_HEADER_TO_PAYLOAD, offset + size); + if lz4 { + let stored = u64_at(&bytes, start) + 1; + bytes[start..start + 8].copy_from_slice(&stored.to_le_bytes()); + } else { + bytes[end - 8..end].fill(0); + } + std::fs::write(path, bytes).unwrap(); + return; + } + offset += (size + 7) & !7; + } + panic!("no compressed DATA object"); +} diff --git a/tests/interoperability/run_reader_hardening.py b/tests/interoperability/run_reader_hardening.py index eaeff1ea..0cc03c28 100755 --- a/tests/interoperability/run_reader_hardening.py +++ b/tests/interoperability/run_reader_hardening.py @@ -122,6 +122,36 @@ def skippable(length: int = 0) -> bytes: return b"\x50\x2a\x4d\x18" + length.to_bytes(4, "little") + bytes(length) +def parse_export(stream: bytes) -> list[dict]: + """Parse journalctl export output into expected_rows() shape. + + Address fields (`__*`) and the header-derived `_BOOT_ID` are not entry + payloads, so they are left out of the comparison. + """ + rows, payloads, realtime, at = [], [], None, 0 + while at < len(stream): + end = stream.index(b"\n", at) + line = stream[at:end] + at = end + 1 + if not line: + rows.append({"realtime": realtime, "payloads": sorted(payloads)}) + payloads, realtime = [], None + continue + if b"=" in line: + name, value = line.split(b"=", 1) + else: + name, size = line, int.from_bytes(stream[at : at + 8], "little") + value = stream[at + 8 : at + 8 + size] + at += 8 + size + 1 + if name == b"__REALTIME_TIMESTAMP": + realtime = int(value) + elif not name.startswith(b"__") and name != b"_BOOT_ID": + payloads.append((name + b"=" + value).hex()) + if payloads or realtime is not None: + rows.append({"realtime": realtime, "payloads": sorted(payloads)}) + return rows + + def expected_rows(entries: list[dict]) -> list[dict]: return [ { @@ -278,6 +308,19 @@ def stock_verify(path: Path, expected_error: bool = False) -> None: f"stock verification disagreement: {path.name}: {result.stderr[-2000:]}" ) + def stock_rows(path: Path, expected: list) -> None: + command = [str(oracle), "--file", str(path), "-o", "export", "--all"] + print(f"+ {shlex.join(command)}", flush=True) + result = subprocess.run( + command, capture_output=True, env=env, timeout=15, check=False # nosemgrep + ) # nosec B603 + got = parse_export(result.stdout) + if result.returncode or got != expected: + raise RuntimeError( + f"stock read disagreement: {path.name}: rc={result.returncode} " + f"rows={len(got)}/{len(expected)} {result.stderr[-2000:]!r}" + ) + zstd = Zstd() if "streams" in selected else None for compact in (False, True): folder = out / ("compact" if compact else "regular") @@ -503,6 +546,12 @@ def stock_verify(path: Path, expected_error: bool = False) -> None: "oversized": (oversized_zstd_frame(768 * 1024 * 1024 + 1), True), "trailing-garbage": (frame + b"garbage", True), } + damaged_message_rows = [dict(row) for row in rows] + damaged_message_rows[0]["payloads"] = [ + payload + for payload in rows[0]["payloads"] + if not bytes.fromhex(payload).startswith(b"MESSAGE=") + ] for name, (stream, error) in streams.items(): padding = end - start - len(stream) - 8 if padding < 0: @@ -515,9 +564,13 @@ def stock_verify(path: Path, expected_error: bool = False) -> None: mutated[start:end] = stream + skippable(padding) path = folder / f"zstd-{name}.journal" path.write_bytes(mutated) - probe(path, "forward", expected=None if error else rows, error=error) + # A damaged stream costs its field, as in stock v262; strict + # verification still rejects the file. + expected = damaged_message_rows if error else rows + probe(path, "forward", expected=expected) probe(path, "verify", error=error) stock_verify(path, error) + stock_rows(path, expected) split = len(data) // 2 whole = ( zstd.frame(data[:split], known=False)