feat: wire FileCheckpointStore into pipeline — persist and restore file offsets (#588)

[strawgate]

Summary

Completes Phase 5c — durable checkpoint persistence with fingerprint-keyed restore.

Design: Checkpoints are keyed by SourceId (file fingerprint), not path. This eliminates the PathUpdate channel message, source_paths tracking, and store.load() fallback — matching how shippers handle checkpoints in practice.

Changes

File	Change
logfwd-io/src/tail.rs	Add set_offset_by_source(SourceId, u64)
logfwd-io/src/input.rs	Add to trait, remove dead source_paths/set_offset
logfwd-io/src/framed.rs	Delegate, remove dead methods
logfwd/src/pipeline.rs	Wire checkpoint store, remove PathUpdate/source_paths
logfwd-test-utils/src/	CountingSink, append_json_lines

Removed

• ChannelMsg::PathUpdate variant
• source_paths: HashMap<SourceId, PathBuf> field
• known_sources tracking in input loop
• source_paths() / set_offset() / file_paths() (dead after PathUpdate removal)

Tests

• test_checkpoint_persisted_after_clean_shutdown
• test_pipeline_resumes_from_checkpoint
• test_channel_msg_data_carries_checkpoints

Closes #588, closes #787

Test plan

• cargo clippy -p logfwd -p logfwd-io -- -D warnings — clean
• cargo fmt --check — clean
• Checkpoint tests pass
• CI (async test flakes are pre-existing, not checkpoint-related)

🤖 Generated with Claude Code

[coderabbitai]

Warning

Rate limit exceeded

@strawgate has exceeded the limit for the number of commits that can be reviewed per hour. Please wait 1 minutes and 32 seconds before requesting another review.

Your organization is not enrolled in usage-based pricing. Contact your admin to enable usage-based pricing to continue reviews beyond the rate limit, or try again in 1 minutes and 32 seconds.

⌛ How to resolve this issue?

After the wait time has elapsed, a review can be triggered using the @coderabbitai review command as a PR comment. Alternatively, push new commits to this PR.

We recommend that you space out your commits to avoid hitting the rate limit.

🚦 How do rate limits work?

CodeRabbit enforces hourly rate limits for each developer per organization.

Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout.

Please see our FAQ for further information.

ℹ️ Review info

⚙️ Run configuration

Configuration used: Repository YAML (base), Organization UI (inherited)

Review profile: ASSERTIVE

Plan: Pro

Run ID: ad779ab5-1368-4eaa-a791-aaab925d2b4b

📥 Commits

Reviewing files that changed from the base of the PR and between 0d22876 and 958d18f.

📒 Files selected for processing (6)

• crates/logfwd-io/src/framed.rs
• crates/logfwd-io/src/input.rs
• crates/logfwd-io/src/tail.rs
• crates/logfwd-test-utils/src/lib.rs
• crates/logfwd-test-utils/src/sinks.rs
• crates/logfwd/src/pipeline.rs

Walkthrough

This change adds durable checkpoint persistence and offset restore by source fingerprint: Pipeline now opens a FileCheckpointStore at startup, loads saved SourceCheckpoints, and restores per-input offsets via a new set_offset_by_source API implemented through FramedInput → FileInput → FileTailer. ChannelMsg::PathUpdate and related path-tracking were removed. Checkpoint updates are persisted on ACK advancement (throttled flush) and a final flush is performed on shutdown. Test utilities CountingSink and append_json_lines were added for tests, and pipeline tests were updated to verify checkpoint creation and resume behavior.

Possibly related PRs

• strawgate/memagent PR 637: Changes to checkpoint wiring and IO/checkpoint APIs across pipeline.rs, framed.rs, input.rs, and tail.rs.
• strawgate/memagent PR 581: Introduces FramedInput/InputSource framing and related IO trait surface that the new set_offset_by_source integrates with.
• strawgate/memagent PR 425: Adds the PipelineMachine/ack receipt machinery (apply_ack/advance semantics) that the checkpoint persistence now hooks into.

---

Caution

Pre-merge checks failed

Please resolve all errors before merging. Addressing warnings is optional.

• Ignore

❌ Failed checks (1 error, 2 warnings)

Check name	Status	Explanation	Resolution
High-Quality Rust Practices	❌ Error	PR contains inaccurate SAFETY comments in checkpoint tests claiming single-threaded execution while code spawns threads that race on global env variables, plus unnecessary clone in hot-path poll() and missing doc comment on CountingSink::new.	Fix SAFETY comments using serial_test crate or remove unsafe blocks; eliminate unnecessary watch_paths.clone() in poll(); add doc comment to CountingSink::new with parameter descriptions.
Documentation Thoroughly Updated	⚠️ Warning	PR implements architectural changes but lacks required documentation updates across five areas: PHASES.md, ARCHITECTURE.md, DESIGN.md, DEVELOPING.md, and three public methods missing doc comments.	Update PHASES.md task 5c status; add FileCheckpointStore section to ARCHITECTURE.md; add ADR to DESIGN.md on fingerprint-keyed restore; add env-var race condition warning to DEVELOPING.md; add doc comments to three public methods.
Maintainer Fitness	⚠️ Warning	PR fails maintainer fitness criteria: missing benchmark results for hot-path changes, lacks risk surface documentation, PHASES.md not updated, and scope lacks justification.	Document benchmark results for ack_all_tickets changes, add risk surface section, update PHASES.md, and consider splitting infrastructure/docs into separate PRs.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Linked Issues check	✅ Passed	PR successfully implements both linked issues: #588 (FileCheckpointStore wiring with startup restore) and #787 (fingerprint-keyed restore eliminating PathUpdate/source_paths). All core requirements met.
Out of Scope Changes check	✅ Passed	All changes directly support checkpoint persistence and fingerprint-keyed restore. No unrelated modifications detected.
Formal Verification Coverage	✅ Passed	PR changes are confined to logfwd-io, logfwd-test-utils, and pipeline.rs consumer logic—all outside the proven logfwd-core kernel. New public functions are test utilities or I/O delegations under explicit exemptions.
Crate Boundary And Dependency Integrity	✅ Passed	PR maintains crate boundary and dependency integrity with no new crates, no new external dependencies, and respects existing layering constraints.

---

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[strawgate]

@coderabbitai please do a thorough review

[coderabbitai]

Tip

For best results, initiate chat on the files or code changes.

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Get the PR diff
gh pr view 784 --repo strawgate/memagent --json files,additions,deletions,changedFiles
echo "---FILES---"
gh pr diff 784 --repo strawgate/memagent

Length of output: 17577

---

🏁 Script executed:

#!/bin/bash
# Find and read FileCheckpointStore implementation
fd -e rs | xargs grep -l "FileCheckpointStore\|CheckpointStore" 2>/dev/null

echo "=== Find checkpoint module ==="
fd "checkpoint" --type f

echo "=== Find pipeline lifecycle ==="
fd "lifecycle.rs" --type f
fd "registry.rs" --type f
fd "batch.rs" --type f

Length of output: 388

---

🏁 Script executed:

#!/bin/bash
echo "=== checkpoint.rs ==="
cat crates/logfwd-io/src/checkpoint.rs

echo ""
echo "=== lifecycle.rs ==="
cat crates/logfwd-core/src/pipeline/lifecycle.rs

echo ""
echo "=== registry.rs ==="
cat crates/logfwd-core/src/pipeline/registry.rs

echo ""
echo "=== pipeline.rs (full, for context) ==="
cat crates/logfwd/src/pipeline.rs

Length of output: 190818

---

@strawgate Here is a thorough review of PR #784.

---

Overall Assessment

The core checkpoint machinery (FileCheckpointStore, PipelineMachine integration, shutdown flush) is solid. The atomic write, ordered-ACK wiring, and final_checkpoints() drain path are all correct. However there are several issues — two of them are correctness bugs that should be fixed before merge.

---

🔴 Critical

1. Test env-var mutation is not thread-safe

Both new tests use unsafe { std::env::set_var("LOGFWD_DATA_DIR", ...) } and later remove_var. The SAFETY comments claim "single-threaded test; no concurrent env reads", but:

• Each test spawns a std::thread internally, so the process is already multi-threaded.
• cargo test runs #[test] items in parallel by default. Any other test that calls from_config → default_data_dir() concurrently will read a polluted or missing env var, producing a data race on process state.
• The two checkpoint tests set the same key — if they run concurrently they corrupt each other's temp dir.

// test_checkpoint_persisted_after_clean_shutdown AND
// test_pipeline_resumes_from_checkpoint both do:
unsafe { std::env::set_var("LOGFWD_DATA_DIR", dir.path()); }
// ... pipeline runs with internal threads ...
unsafe { std::env::remove_var("LOGFWD_DATA_DIR"); }

Fix: add serial_test (or a similar crate) and annotate both tests with #[serial], or pass the data dir explicitly through a new from_config_with_data_dir(…, Option<&Path>) overload so tests never touch env vars.

---

2. PathUpdate ordering hole — checkpoint silently discarded on resume

In ack_all_tickets, when a checkpoint advances the store is updated with:

store.update(SourceCheckpoint {
    source_id: advance.source.0,
    path: self.source_paths.get(&advance.source).cloned(), // ← can be None
    offset,
});

If the PathUpdate message for this SourceId has not yet been processed (a real possibility under any message-ordering jitter), source_paths does not contain the key and path is stored as None.

On the next startup the restore loop silently ignores path: None checkpoints:

for cp in &saved_checkpoints {
    if let Some(path) = &cp.path {   // ← skipped when path is None
        for input in &mut inputs {
            input.source.set_offset(path, cp.offset);
        }
    }
}

The file is re-read from the beginning — the very bug this PR exists to fix. The PR description correctly notes "Must arrive before the first Data referencing a new SourceId" but there is no enforcement.

Fix options:

• Buffer Data messages that reference unknown SourceIds until the corresponding PathUpdate arrives, or
• Emit PathUpdate atomically with the first Data message, making them part of a single channel message, or
• Fall back to source_id-only restore (look up path from the live input's source_paths() at restore time, keyed by fingerprint).

---

🟡 Significant

3. Flush-throttle broken on persistent write errors

In ack_all_tickets, last_checkpoint_flush is only reset on success:

if let Err(e) = store.flush() {
    eprintln!("pipeline: checkpoint flush error: {e}");
} else {
    self.last_checkpoint_flush = Instant::now(); // ← not reset on error
}

After the first flush failure (e.g. disk full) last_checkpoint_flush.elapsed() grows indefinitely. Every subsequent ack_all_tickets call will attempt a flush, causing a flush storm — the opposite of what the 5-second throttle is meant to prevent.

Fix: reset the timer unconditionally (or use exponential back-off):

// reset regardless of outcome so the 5-second window is preserved
self.last_checkpoint_flush = Instant::now();
if let Err(e) = store.flush() {
    eprintln!("pipeline: checkpoint flush error: {e}");
}

---

4. checkpoints.json not namespaced — multi-pipeline configs corrupt each other

FileCheckpointStore::open(default_data_dir()) is called unconditionally for every pipeline. When a config defines multiple pipelines they all write to the same {data_dir}/checkpoints.json. The second flush() overwrites the first pipeline's checkpoints.

Fix: namespace by pipeline name, e.g. {data_dir}/{pipeline_name}/checkpoints.json.

---

5. No directory fsync after atomic rename

The flush() implementation correctly fsyncs the tmp file before renaming, but does not fsync the parent directory after the rename. On crash between rename and the directory entry being flushed (possible on ext4 without data=ordered, XFS, btrfs) the rename may be lost and the file reverts to its pre-rename state.

std::fs::rename(&tmp, &final_path)?;
// Missing: fsync the parent directory

Fix:

std::fs::rename(&tmp, &final_path)?;
// Sync the directory so the new dirent is durable
let dir = std::fs::File::open(&self.data_dir)?;
dir.sync_all()?;

---

6. Checkpoint store opened unnecessarily for non-file inputs

from_config opens FileCheckpointStore for all pipeline types (generator, OTLP, UDP, TCP). Those sources have no file offset to checkpoint. The store is created, checkpoints are loaded (none match), and the file is written on shutdown with path: None entries — wasted I/O and a potential confusion vector.

Fix: open the store only when at least one InputType::File is present in config.inputs.

---

🔵 Minor / Polish

7. SAFETY comments are inaccurate

Both new tests spawn threads and state "single-threaded test; no concurrent env reads". The comment should be removed or rewritten to accurately describe the actual risk (cross-test parallelism, not in-test concurrency).

8. append_json_lines counts existing lines by reading the full file

let existing = std::fs::read_to_string(path).unwrap_or_default();
let start_seq = existing.lines().count();

For large files this reads the entire content into memory just to count newlines. A cheaper alternative is seeking to the end and tracking byte offset, or accepting start_seq as a parameter (making the function more explicit about its contract).

9. CountingSink uses Ordering::Relaxed — document the rationale

Relaxed ordering is correct here (no synchronization guarantees are needed beyond the atomic increment itself), but adding a comment would help readers who might question the choice.

10. Missing benchmark results (per repo policy)

Per the project's coding guidelines, hot-path changes require before/after benchmark results from just bench. ack_all_tickets is called on every batch and now performs additional work (source_paths lookup, store.update, conditional store.flush). Please include benchmark numbers.

---

Summary Table

#	Severity	File	Description
1	🔴 Critical	pipeline.rs tests	Env-var race across parallel test threads
2	🔴 Critical	pipeline.rs	path: None checkpoint silently ignored on restore
3	🟡 Significant	pipeline.rs	Flush storm on persistent write errors
4	🟡 Significant	pipeline.rs	Single checkpoints.json clobbers multi-pipeline configs
5	🟡 Significant	checkpoint.rs	Missing directory fsync after atomic rename
6	🟡 Significant	pipeline.rs	Checkpoint store opened for non-file inputs
7	🔵 Minor	pipeline.rs tests	Inaccurate SAFETY comments
8	🔵 Minor	test-utils/lib.rs	append_json_lines reads entire file to count lines
9	🔵 Minor	test-utils/sinks.rs	Ordering::Relaxed undocumented
10	🔵 Minor	pipeline.rs	Missing just bench results for hot path

Items 1 and 2 are correctness bugs that affect the primary feature this PR delivers. Items 3–6 should be addressed before merge. The rest can be follow-ups if preferred.

---

🧠 Learnings used

Learnt from: CR
Repo: strawgate/memagent PR: 0
File: coderabbit-custom-pre-merge-checks-unique-id-file-non-traceable-F7F2B60C-1728-4C9A-8889-4F2235E186CA.txt:0-0
Timestamp: 2026-04-03T02:15:17.278Z
Learning: Applies to src/crate/logfwd-core/**/pipeline/lifecycle.rs : Any change to pipeline/lifecycle.rs, PipelineMachine state transitions, drain protocol, checkpoint ordering, batch sequence numbering, or ordered-ACK mechanism MUST address TLA+ coverage

---

Learnt from: CR
Repo: strawgate/memagent PR: 0
File: coderabbit-custom-pre-merge-checks-unique-id-file-non-traceable-F7F2B60C-1728-4C9A-8889-4F2235E186CA.txt:0-0
Timestamp: 2026-04-03T02:15:17.278Z
Learning: Applies to src/crate/logfwd-core/**/pipeline/lifecycle.rs : State machine lifecycle.rs MUST prove all valid state transitions preserve invariants: no batch creation after drain begins, committed checkpoint monotonically non-decreasing, stopped state is terminal

---

Learnt from: CR
Repo: strawgate/memagent PR: 0
File: coderabbit-custom-pre-merge-checks-unique-id-file-non-traceable-F7F2B60C-1728-4C9A-8889-4F2235E186CA.txt:0-0
Timestamp: 2026-04-03T02:15:17.278Z
Learning: Applies to src/crate/logfwd-core/**/pipeline/**/*.rs : New async pipeline code MUST have proptest tests for: event sequences in arbitrary order, acks arriving in different order than submits, drain requested while batches in-flight

---

Learnt from: CR
Repo: strawgate/memagent PR: 0
File: coderabbit-custom-pre-merge-checks-unique-id-file-non-traceable-F7F2B60C-1728-4C9A-8889-4F2235E186CA.txt:0-0
Timestamp: 2026-04-03T02:15:17.278Z
Learning: Applies to tla/PipelineMachine.tla : TLA+ spec must verify safety invariants: TypeOK, NoDoubleComplete, DrainCompleteness, CheckpointOrderingInvariant, CommittedNeverAheadOfAcked, InFlightImpliesCreated, AckedImpliesCreated with PipelineMachine.cfg

---

Learnt from: CR
Repo: strawgate/memagent PR: 0
File: coderabbit-custom-pre-merge-checks-unique-id-file-non-traceable-F7F2B60C-1728-4C9A-8889-4F2235E186CA.txt:0-0
Timestamp: 2026-04-03T02:15:17.278Z
Learning: Applies to src/crate/logfwd-core/**/*.rs : Hot-path code changes include before/after benchmark results from `just bench`

---

Learnt from: CR
Repo: strawgate/memagent PR: 0
File: coderabbit-custom-pre-merge-checks-unique-id-file-non-traceable-F7F2B60C-1728-4C9A-8889-4F2235E186CA.txt:0-0
Timestamp: 2026-04-03T02:15:17.278Z
Learning: Applies to src/crate/logfwd-core/**/pipeline/registry.rs : State machine registry.rs MUST prove: sequence numbers are strictly monotonic, no duplicate batch IDs

---

Learnt from: CR
Repo: strawgate/memagent PR: 0
File: coderabbit-custom-pre-merge-checks-unique-id-file-non-traceable-F7F2B60C-1728-4C9A-8889-4F2235E186CA.txt:0-0
Timestamp: 2026-04-03T02:15:17.278Z
Learning: Applies to src/crate/logfwd-core/**/pipeline/batch.rs : State machine batch.rs MUST prove typestate: Queued → Sending is only allowed first transition, Sending → Acked and Sending → Rejected are only allowed completions

[coderabbitai]

Actionable comments posted: 2

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@crates/logfwd-test-utils/src/lib.rs`:
- Around line 36-56: The docstring for append_json_lines misleadingly refers to
`start_seq` as if it's a parameter; update the comment to state that sequence
IDs start from the current line count of the target file (computed by reading
the file) rather than an input parameter. Mention that the function auto-detects
the starting sequence by counting existing lines and then appends `count` NDJSON
lines with IDs from that computed `start_seq`, and remove or reword the phrase
implying `start_seq` is provided by the caller.

In `@crates/logfwd/src/pipeline.rs`:
- Around line 2654-2657: Update the misleading SAFETY comments around the unsafe
std::env::set_var("LOGFWD_DATA_DIR", ...) block and the other comment near the
"single-threaded test": state that CHECKPOINT_ENV_MUTEX serializes access to
environment variable mutation for this test, and clarify that spawned threads
(e.g., the metrics thread started later) do not read or rely on environment
variables rather than claiming there are "no concurrent env reads" or that the
test is "single-threaded"; reference the unsafe block around set_var and the
comment near the later test to make these clarifications.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Repository YAML (base), Organization UI (inherited)

Review profile: ASSERTIVE

Plan: Pro

Run ID: 2afaa399-a22a-4a94-a9d1-c2245c46d31c

📥 Commits

Reviewing files that changed from the base of the PR and between 0635dc2 and c98db9c.

📒 Files selected for processing (5)

• crates/logfwd-io/src/checkpoint.rs
• crates/logfwd-io/src/tcp_input.rs
• crates/logfwd-test-utils/src/lib.rs
• crates/logfwd-test-utils/src/sinks.rs
• crates/logfwd/src/pipeline.rs

[coderabbitai]

Actionable comments posted: 1

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)

crates/logfwd/src/pipeline.rs (1)

886-923: 🧹 Nitpick | 🔵 Trivial

Consider extracting PathUpdate-before-Data logic.

Lines 895-915 duplicate the send pattern from lines 843-871. A helper like send_data_with_path_update(...) could reduce duplication.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@crates/logfwd/src/pipeline.rs` around lines 886 - 923, Extract the duplicated
"send PathUpdate before Data" logic into a helper (e.g.,
send_data_with_path_update) that accepts the source (or input), tx, metrics,
known_sources and the data/checkpoints to send; move the block that computes
current_sources, new_sources, filters source_paths and calls
blocking_send_channel_msg with ChannelMsg::PathUpdate into that helper and then
have it send ChannelMsg::Data (using blocking_send_channel_msg) so both the
earlier occurrence (lines ~843-871) and this shutdown-path (lines ~895-915) call
the same function, reducing duplication and keeping calls to
input.source.checkpoint_data(), input.source.source_paths(),
blocking_send_channel_msg, ChannelMsg::PathUpdate and ChannelMsg::Data
centralized.

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@crates/logfwd/src/pipeline.rs`:
- Around line 723-745: The code repeatedly calls store.load(advance.source.0) as
a fallback when self.source_paths lacks an entry; cache the persisted path on
first load to avoid repeated I/O/lookup. Change the fallback flow in the
machine.apply_ack handling: when you call store.load(...) and get a
SourceCheckpoint with a path, insert that path into self.source_paths (or a
dedicated persisted-path cache) keyed by advance.source so subsequent
checkpoints use the in-memory path instead of calling store.load again; ensure
you still prefer PathUpdate entries from self.source_paths and only
consult/update the cache when store.load returns a path.

---

Outside diff comments:
In `@crates/logfwd/src/pipeline.rs`:
- Around line 886-923: Extract the duplicated "send PathUpdate before Data"
logic into a helper (e.g., send_data_with_path_update) that accepts the source
(or input), tx, metrics, known_sources and the data/checkpoints to send; move
the block that computes current_sources, new_sources, filters source_paths and
calls blocking_send_channel_msg with ChannelMsg::PathUpdate into that helper and
then have it send ChannelMsg::Data (using blocking_send_channel_msg) so both the
earlier occurrence (lines ~843-871) and this shutdown-path (lines ~895-915) call
the same function, reducing duplication and keeping calls to
input.source.checkpoint_data(), input.source.source_paths(),
blocking_send_channel_msg, ChannelMsg::PathUpdate and ChannelMsg::Data
centralized.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Repository YAML (base), Organization UI (inherited)

Review profile: ASSERTIVE

Plan: Pro

Run ID: 7c42b466-2b12-488a-8fd2-5b0b3423e13a

📥 Commits

Reviewing files that changed from the base of the PR and between c98db9c and 36f2414.

📒 Files selected for processing (2)

• crates/logfwd-test-utils/src/lib.rs
• crates/logfwd/src/pipeline.rs

[coderabbitai]

🧹 Nitpick | 🔵 Trivial

Consider caching previously persisted paths.

store.load(advance.source.0) inside the ticket loop performs I/O (or at least a hash lookup) on every advancing checkpoint. If source_paths frequently lacks entries, this fallback runs repeatedly.

Consider caching the path on first load or ensuring PathUpdate always arrives before any Data referencing that source.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@crates/logfwd/src/pipeline.rs` around lines 723 - 745, The code repeatedly
calls store.load(advance.source.0) as a fallback when self.source_paths lacks an
entry; cache the persisted path on first load to avoid repeated I/O/lookup.
Change the fallback flow in the machine.apply_ack handling: when you call
store.load(...) and get a SourceCheckpoint with a path, insert that path into
self.source_paths (or a dedicated persisted-path cache) keyed by advance.source
so subsequent checkpoints use the in-memory path instead of calling store.load
again; ensure you still prefer PathUpdate entries from self.source_paths and
only consult/update the cache when store.load returns a path.

[strawgate]

@coderabbitai please do a thorough code review

[coderabbitai]

@strawgate Sure — I'll kick off a full re-review of PR #784 now, taking into account all the latest commits (including the fixes to the SAFETY comments and the directory fsync).

---

🧠 Learnings used

Learnt from: CR
Repo: strawgate/memagent PR: 0
File: coderabbit-custom-pre-merge-checks-unique-id-file-non-traceable-F7F2B60C-1728-4C9A-8889-4F2235E186CA.txt:0-0
Timestamp: 2026-04-03T03:25:32.419Z
Learning: Applies to logfwd-core/src/pipeline/lifecycle.rs : Any PR modifying pipeline/lifecycle.rs, PipelineMachine state transitions, drain protocol, checkpoint ordering, batch sequence numbering, or ordered-ACK mechanism MUST address TLA+ coverage

---

Learnt from: CR
Repo: strawgate/memagent PR: 0
File: coderabbit-custom-pre-merge-checks-unique-id-file-non-traceable-F7F2B60C-1728-4C9A-8889-4F2235E186CA.txt:0-0
Timestamp: 2026-04-03T03:25:32.419Z
Learning: Applies to logfwd-core/src/pipeline/**/*.rs : New async pipeline code MUST have proptest tests for event sequences arriving in arbitrary order, acks arriving in different order than submits, and drain requested while batches in-flight

---

Learnt from: CR
Repo: strawgate/memagent PR: 0
File: coderabbit-custom-pre-merge-checks-unique-id-file-non-traceable-F7F2B60C-1728-4C9A-8889-4F2235E186CA.txt:0-0
Timestamp: 2026-04-03T03:25:32.419Z
Learning: Applies to logfwd-core/src/pipeline/lifecycle.rs : Pipeline lifecycle proofs must verify: no batch creation after drain begins, committed checkpoint is monotonically non-decreasing, stopped state is terminal

---

Learnt from: CR
Repo: strawgate/memagent PR: 0
File: coderabbit-custom-pre-merge-checks-unique-id-file-non-traceable-F7F2B60C-1728-4C9A-8889-4F2235E186CA.txt:0-0
Timestamp: 2026-04-03T03:25:32.419Z
Learning: TLA+ PR requirements: provide explicit confirmation that changes do not affect TLA+ properties with reasoning, OR update TLA+ spec and show all properties still hold under both PipelineMachine.cfg and PipelineMachine.liveness.cfg

---

Learnt from: CR
Repo: strawgate/memagent PR: 0
File: coderabbit-custom-pre-merge-checks-unique-id-file-non-traceable-F7F2B60C-1728-4C9A-8889-4F2235E186CA.txt:0-0
Timestamp: 2026-04-03T03:25:32.419Z
Learning: Applies to logfwd-core/src/pipeline/{lifecycle,batch,registry}.rs : State machine transitions (pipeline/lifecycle.rs, pipeline/batch.rs, pipeline/registry.rs) MUST prove all valid state transitions preserve invariants

---

Learnt from: CR
Repo: strawgate/memagent PR: 0
File: coderabbit-custom-pre-merge-checks-unique-id-file-non-traceable-F7F2B60C-1728-4C9A-8889-4F2235E186CA.txt:0-0
Timestamp: 2026-04-03T03:25:32.419Z
Learning: Applies to logfwd-core/src/{json_scanner,scan_config,cri}.rs : New scanner behavior (field extraction, type detection, duplicate key handling, CRI parsing) MUST have proptest coverage for escape sequences crossing 64-byte SIMD boundaries, fields in different orders, and duplicate key handling

---

Learnt from: CR
Repo: strawgate/memagent PR: 0
File: coderabbit-custom-pre-merge-checks-unique-id-file-non-traceable-F7F2B60C-1728-4C9A-8889-4F2235E186CA.txt:0-0
Timestamp: 2026-04-03T03:25:32.419Z
Learning: Applies to {logfwd-arrow,logfwd-io,logfwd-transform,logfwd-output,logfwd-config}/src/**/*.rs : All unsafe blocks outside logfwd-core must have a // SAFETY: comment stating the specific invariant

---

Learnt from: CR
Repo: strawgate/memagent PR: 0
File: coderabbit-custom-pre-merge-checks-unique-id-file-non-traceable-F7F2B60C-1728-4C9A-8889-4F2235E186CA.txt:0-0
Timestamp: 2026-04-03T03:25:32.419Z
Learning: Applies to logfwd-core/src/pipeline/batch.rs : Pipeline batch typestate proofs must verify: Queued → Sending is the only allowed first transition, Sending → Acked and Sending → Rejected are the only allowed completions

---

Learnt from: CR
Repo: strawgate/memagent PR: 0
File: coderabbit-custom-pre-merge-checks-unique-id-file-non-traceable-F7F2B60C-1728-4C9A-8889-4F2235E186CA.txt:0-0
Timestamp: 2026-04-03T03:25:32.419Z
Learning: Applies to logfwd-core/src/{framer,cri,json_scanner,scan_config,aggregator}.rs : Parser functions SHOULD have a correctness proof comparing output against an independently-computed oracle (not the production code rewritten)

---

Learnt from: CR
Repo: strawgate/memagent PR: 0
File: coderabbit-custom-pre-merge-checks-unique-id-file-non-traceable-F7F2B60C-1728-4C9A-8889-4F2235E186CA.txt:0-0
Timestamp: 2026-04-03T03:25:32.419Z
Learning: Applies to logfwd-core/src/pipeline/registry.rs : Pipeline registry proofs must verify: sequence numbers are strictly monotonic, no duplicate batch IDs

✅ Actions performed

Full review triggered.

[strawgate]

Reviewed: all CodeRabbit nitpicks resolved. The remaining caching suggestion is a valid optimization but not blocking. Branch is mergeable and up to date with master. Ready to merge.

[coderabbitai]

Actionable comments posted: 4

♻️ Duplicate comments (1)

crates/logfwd-test-utils/src/lib.rs (1)

36-38: ⚠️ Potential issue | 🟡 Minor

Fix the stale start_seq doc comment.

start_seq is derived inside append_json_lines; it is not a caller-supplied input.

📝 Suggested doc fix

-/// Append `count` NDJSON lines to an existing file.
-/// Sequence IDs start from `start_seq`. Used by resume tests to simulate
-/// new data appearing after the initial batch.
+/// Append `count` NDJSON lines to an existing file.
+/// Sequence IDs continue from the current line count. Used by resume tests
+/// to simulate new data appearing after the initial batch.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@crates/logfwd-test-utils/src/lib.rs` around lines 36 - 38, The doc comment
incorrectly states that `start_seq` is a caller-supplied parameter; update the
comment for the function append_json_lines to explain that sequence IDs are
computed within the function (derived from existing file contents or an internal
counter) rather than passed in by the caller, and remove or rephrase any wording
implying `start_seq` is an external argument so the docs accurately reflect the
function's behavior.

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@crates/logfwd/src/pipeline.rs`:
- Around line 199-205: The restore loop currently only uses the unstable
fingerprint SourceId (via set_offset_by_source) which breaks when files grow;
change checkpoint persistence to store the canonical file path (not None) and in
the restore in pipeline.rs iterate saved_checkpoints and first try to restore by
that canonical path (e.g. call the source API to set offset by path, use
something like input.source.set_offset_by_path(canonical_path, cp.offset)), then
only if that fails fall back to the fingerprint-based restore using SourceId and
set_offset_by_source(SourceId(cp.source_id), cp.offset); update the checkpoint
write path to persist canonical_path for each checkpoint so the primary restore
key is the stable path and fingerprint remains a fallback (references:
saved_checkpoints, cp.source_id, cp.offset, SourceId, inputs,
set_offset_by_source, set_offset_by_path).
- Around line 157-169: Pipeline::from_config currently calls
FileCheckpointStore::open(default_data_dir()) which causes all pipelines to
share the same checkpoints.json and allows one pipeline's
FileCheckpointStore::flush to overwrite another's in-memory map; change this to
open a pipeline-scoped store (e.g. default_data_dir().join(pipeline_name) or
default_data_dir().join(format!("checkpoints-{}.json", name))) so each pipeline
gets its own file/dir. Specifically, update the call site in
Pipeline::from_config to pass a per-pipeline path/namespace into
FileCheckpointStore::open (or add a new FileCheckpointStore::open_at(path) if
needed), and ensure FileCheckpointStore::load_all and FileCheckpointStore::flush
read/write that per-pipeline path so checkpoints are namespaced by the pipeline
`name` (or other stable pipeline identity).
- Around line 704-712: The current logic in the flush throttle uses
last_checkpoint_flush and only updates it on successful store.flush(), so after
the first flush error the timestamp never advances and the code repeatedly
retries and floods logs; modify the block around any_advanced /
self.checkpoint_store so that last_checkpoint_flush = Instant::now() is set
regardless of whether store.flush() returns Err or Ok (or implement an
increasing backoff), i.e., move the last_checkpoint_flush update out of the
success branch (or update on error with appropriate backoff) while keeping the
existing error logging for checkpoint_store.flush().
- Around line 2527-2531: Tests mutate the shared env var LOGFWD_DATA_DIR causing
races; either add the serial_test crate and mark the affected tests with
#[serial] or refactor the tested constructors to accept a data_dir so tests
don't set env globals. To apply the serial approach: add serial_test to
Cargo.toml, import the serial attribute (use serial_test::serial) and annotate
test_checkpoint_persisted_after_clean_shutdown,
test_pipeline_resumes_from_checkpoint and the two other tests that set
LOGFWD_DATA_DIR with #[serial]. Alternatively, implement a
from_config/from_data_dir constructor for the Pipeline (or relevant builder used
in those tests) and update the tests to instantiate Pipeline with an explicit
data_dir instead of calling std::env::set_var("LOGFWD_DATA_DIR").

---

Duplicate comments:
In `@crates/logfwd-test-utils/src/lib.rs`:
- Around line 36-38: The doc comment incorrectly states that `start_seq` is a
caller-supplied parameter; update the comment for the function append_json_lines
to explain that sequence IDs are computed within the function (derived from
existing file contents or an internal counter) rather than passed in by the
caller, and remove or rephrase any wording implying `start_seq` is an external
argument so the docs accurately reflect the function's behavior.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Repository YAML (base), Organization UI (inherited)

Review profile: ASSERTIVE

Plan: Pro

Run ID: a60c3ca2-a016-45be-b4c2-7273189ceee2

📥 Commits

Reviewing files that changed from the base of the PR and between 36f2414 and f6c9aa8.

📒 Files selected for processing (6)

• crates/logfwd-io/src/framed.rs
• crates/logfwd-io/src/input.rs
• crates/logfwd-io/src/tail.rs
• crates/logfwd-test-utils/src/lib.rs
• crates/logfwd-test-utils/src/sinks.rs
• crates/logfwd/src/pipeline.rs

[coderabbitai]

⚠️ Potential issue | 🔴 Critical

Fingerprint-only restore is not stable across restarts.

set_offset_by_source() is now the only restore path, and both persistence paths write path: None. Here SourceId comes from the tailer's fingerprint of the first fingerprint_bytes bytes, so a file checkpointed while smaller than that window stops matching after it grows before the next start; restore silently no-ops and the file replays from offset 0. Persist the canonical path and use it as the primary restore key, or switch to a source identity that is stable across growth and duplicate prefixes.

Also applies to: 463-468, 695-699

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@crates/logfwd/src/pipeline.rs` around lines 199 - 205, The restore loop
currently only uses the unstable fingerprint SourceId (via set_offset_by_source)
which breaks when files grow; change checkpoint persistence to store the
canonical file path (not None) and in the restore in pipeline.rs iterate
saved_checkpoints and first try to restore by that canonical path (e.g. call the
source API to set offset by path, use something like
input.source.set_offset_by_path(canonical_path, cp.offset)), then only if that
fails fall back to the fingerprint-based restore using SourceId and
set_offset_by_source(SourceId(cp.source_id), cp.offset); update the checkpoint
write path to persist canonical_path for each checkpoint so the primary restore
key is the stable path and fingerprint remains a fallback (references:
saved_checkpoints, cp.source_id, cp.offset, SourceId, inputs,
set_offset_by_source, set_offset_by_path).

[coderabbitai]

⚠️ Potential issue | 🔴 Critical

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
set -euo pipefail
rg -n 'LOGFWD_DATA_DIR|CHECKPOINT_ENV_MUTEX|serial_test'

Repository: strawgate/memagent

Length of output: 1123

---

🏁 Script executed:

#!/bin/bash
set -euo pipefail

# Extract the test functions and their context around the identified lines
echo "=== Extracting test functions around LOGFWD_DATA_DIR mutations ==="

# Get lines around 2527-2531
echo "--- Lines 2520-2535 (first mutation) ---"
sed -n '2520,2535p' crates/logfwd/src/pipeline.rs | cat -n

echo ""
echo "--- Lines 2560-2575 (second mutation) ---"
sed -n '2560,2575p' crates/logfwd/src/pipeline.rs | cat -n

echo ""
echo "--- Lines 2573-2590 (third mutation) ---"
sed -n '2573,2590p' crates/logfwd/src/pipeline.rs | cat -n

echo ""
echo "--- Lines 2650-2670 (fourth mutation) ---"
sed -n '2650,2670p' crates/logfwd/src/pipeline.rs | cat -n

Repository: strawgate/memagent

Length of output: 3572

---

🏁 Script executed:

#!/bin/bash
set -euo pipefail

# Check if logfwd has serial_test dependency
echo "=== Checking logfwd crate dependencies ==="
grep -A 20 '^\[dev-dependencies\]' crates/logfwd/Cargo.toml || echo "No [dev-dependencies] section found"

echo ""
echo "=== Checking for any #[serial] usage in pipeline.rs ==="
rg -n '#\[serial\]' crates/logfwd/src/pipeline.rs || echo "No #[serial] attributes found"

echo ""
echo "=== Checking test module attributes in pipeline.rs ==="
sed -n '2500,2530p' crates/logfwd/src/pipeline.rs | cat -n

Repository: strawgate/memagent

Length of output: 1813

---

Serialize environment mutations in tests to prevent race conditions.

Tests test_checkpoint_persisted_after_clean_shutdown (line 2522), test_pipeline_resumes_from_checkpoint (line 2576), and two others mutate the shared process environment (LOGFWD_DATA_DIR) without serialization. Under cargo test parallelism, these calls race with other tests; a panic before cleanup leaks state. The logfwd crate lacks a serial_test dependency, so add it and decorate these tests with #[serial], or refactor to pass data_dir explicitly via a from_config variant.

Applies to lines: 2527–2531, 2568–2570, 2580–2583, 2659–2661.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@crates/logfwd/src/pipeline.rs` around lines 2527 - 2531, Tests mutate the
shared env var LOGFWD_DATA_DIR causing races; either add the serial_test crate
and mark the affected tests with #[serial] or refactor the tested constructors
to accept a data_dir so tests don't set env globals. To apply the serial
approach: add serial_test to Cargo.toml, import the serial attribute (use
serial_test::serial) and annotate
test_checkpoint_persisted_after_clean_shutdown,
test_pipeline_resumes_from_checkpoint and the two other tests that set
LOGFWD_DATA_DIR with #[serial]. Alternatively, implement a
from_config/from_data_dir constructor for the Pipeline (or relevant builder used
in those tests) and update the tests to instantiate Pipeline with an explicit
data_dir instead of calling std::env::set_var("LOGFWD_DATA_DIR").

[strawgate]

CI failing with 2 test failures:

1. test_pipeline_transform_error_skips_batch_continues — expects transform_errors > 0 but gets 0 (pipeline.rs:1340). The checkpoint persistence changes may have altered error propagation behavior.

2. allocation_per_row_decreases_with_scale — allocation scaling test failing, possibly related to the new checkpoint store overhead.

Both need investigation against the new checkpoint persistence logic.

[strawgate]

@coderabbitai can you please fully review this change?

[coderabbitai]

🧩 Analysis chain

@strawgate I'll kick off a full re-review now and also investigate the two CI failures.

---

🏁 Script executed:

#!/bin/bash
echo "=== Investigate test_pipeline_transform_error_skips_batch_continues ==="
rg -n "transform_error_skips_batch\|transform_errors\|TransformError\|skip.*batch\|transform_error" crates/logfwd/src/pipeline.rs -A5 -B5

echo ""
echo "=== Look for the test itself ==="
rg -n "test_pipeline_transform_error_skips_batch_continues" crates/logfwd/src/pipeline.rs -A 60

echo ""
echo "=== Look for transform error handling in pipeline ==="
rg -n "transform_error\|TransformError\|skip_batch\|error_count\|transform_errors" crates/logfwd/src/ -r --type rust -A3 -B3

echo ""
echo "=== Look at ack_all_tickets and checkpoint changes around error propagation ==="
rg -n "ack_all_tickets\|apply_ack\|Rejected\|transform" crates/logfwd/src/pipeline.rs -A5 -B3

Length of output: 2749

---

🏁 Script executed:

#!/bin/bash
echo "=== Full pipeline.rs around transform handling ==="
cat crates/logfwd/src/pipeline.rs | head -200

echo ""
echo "=== Search for transform_errors metric usage ==="
grep -n "transform_errors\|transform_error\|TransformError\|apply_transform\|do_transform" crates/logfwd/src/pipeline.rs

echo ""
echo "=== Look at the run() and polling loop — transform path ==="
grep -n "transform\|Rejected\|skip\|error" crates/logfwd/src/pipeline.rs | head -80

echo ""
echo "=== Look at allocation_per_row test ==="
grep -rn "allocation_per_row_decreases_with_scale" crates/ --include="*.rs" -A 40

echo ""
echo "=== Look at checkpoint store / overhead added ==="
grep -n "checkpoint_store\|FileCheckpointStore\|last_checkpoint_flush\|from_config" crates/logfwd/src/pipeline.rs | head -60

Length of output: 20833

---

🏁 Script executed:

#!/bin/bash
echo "=== flush_batch / send_batch around transform error path ==="
sed -n '489,680p' crates/logfwd/src/pipeline.rs

echo ""
echo "=== ack_all_tickets full function ==="
grep -n "fn ack_all_tickets" crates/logfwd/src/pipeline.rs
sed -n '680,730p' crates/logfwd/src/pipeline.rs

echo ""
echo "=== allocation_scaling.rs full file ==="
cat crates/logfwd/tests/allocation_scaling.rs

echo ""
echo "=== measure_pipeline function ==="
grep -n "fn measure_pipeline\|from_config\|checkpoint_store\|data_dir" crates/logfwd/tests/allocation_scaling.rs -A 5

echo ""
echo "=== run_async / polling loop after checkpoint changes ==="
sed -n '300,490p' crates/logfwd/src/pipeline.rs

echo ""
echo "=== shutdown / drain path ==="
sed -n '420,490p' crates/logfwd/src/pipeline.rs

Length of output: 25724

---

@strawgate Here is the full re-review with CI failure root-cause analysis.

---

CI Failure Analysis

🔴 Failure 1 — test_pipeline_transform_error_skips_batch_continues (line 1340)

Root cause: zero-row batch early-exit bypasses transform.

The test uses SELECT nonexistent_col FROM logs. SqlTransform::scan_config() for this query configures the scanner to extract only nonexistent_col. The input JSON ({"level":"INFO","msg":"hello 0"}) contains no such field, so StreamingSimdScanner::scan() returns a zero-row RecordBatch. The following guard in flush_batch then fires:

// ~line 616
if batch.num_rows() == 0 {
    self.ack_all_tickets(sending, true);
    return;   // ← returns here; transform.execute() is never called
}
// ... inc_transform_error() is below this, unreachable

transform_errors is therefore never incremented regardless of how long the pipeline runs. The transform is not broken — it simply never sees data.

Fix options (in order of preference):

1. Rewrite the test SQL to reference a column that does exist but uses an expression that fails at execute time (e.g. a bad cast or a non-existent UDF):

   SELECT CAST(level AS INT) FROM logs  -- level is a string → DataFusion type error

2. Add a scan-level error test separately for the SELECT nonexistent_col case, asserting scan_errors (not transform_errors) if that is the accurate metric.

3. Change the test input to include nonexistent_col in the JSON so the scanner produces rows, then fail inside the transform via a function:

   // Use SQL that reads a present column but applies an invalid operation

---

🔴 Failure 2 — allocation_per_row_decreases_with_scale

Root cause: per-batch checkpoint allocations inflate measurements non-uniformly.

from_config now opens FileCheckpointStore for every pipeline including the null-output file-input pipeline used by measure_pipeline. During run() (inside the Region), every ack_all_tickets call hits:

store.update(SourceCheckpoint {
    source_id: advance.source.0,
    path: None,
    offset,
});

This allocates a SourceCheckpoint struct (and an internal HashMap/Vec reallocation) on every batch where a checkpoint advances. The allocation scales with batch count, not row count. At 10K rows with a 10 ms batch timeout, more smaller batches may fire before shutdown (the poll interval is 5ms and the file is pre-written — the input thread can drain the file into many fast batches). At 100K or 500K rows the file takes longer to drain; batches are larger and fewer checkpoint updates occur per row, but the absolute count of updates is also higher.

Whether this breaks per_row_100k <= per_row_10k * 1.2 depends on CI timing jitter — consistent with "possibly related to the new checkpoint store overhead".

Fix: Suppress checkpoint persistence in tests that are not testing checkpoints:

// Option A: open store only when env / config explicitly enables it
// (already suggested as Issue `#6` — open only for file inputs, optional flag)

// Option B: add a `from_config` override accepted by Pipeline for tests
pub fn from_config_no_checkpoint(name: &str, config: &PipelineConfig, meter: &Meter, base_path: Option<&Path>) -> Result<Self, String> {
    let mut p = Self::from_config(name, config, meter, base_path)?;
    p.checkpoint_store = None;
    p
}

Or simply: guard FileCheckpointStore::open behind a config flag (enable_checkpoints: bool, default true in production, false in unit/bench tests). The allocation test doesn't need durable checkpoints.

---

Remaining Issues from Prior Review

#	Status	Notes
1 (env-var race)	❓ Unknown	Didn't see serial_test added; please confirm
2 (path:None hole)	✅ Fixed	set_offset_by_source used directly — no path lookup needed on restore
3 (flush-throttle on error)	🔴 Still present	last_checkpoint_flush reset only in else branch of flush error check (line 710); persistent errors cause flush storms
4 (single checkpoints.json)	🟡 Open	All pipelines share {data_dir}/checkpoints.json; multi-pipeline configs will clobber each other
5 (parent dir fsync)	✅ Fixed (per user)
6 (store for non-file inputs)	🟡 Open	Also root cause of allocation test failure
7–9 (minor polish)	✅ Resolved (per user)
10 (bench results)	🟡 Open	ack_all_tickets hot-path now does extra work; benchmark numbers still missing

---

Recommendation

The two CI failures are both caused by the checkpoint persistence code being active in test paths where it shouldn't be. Fixing Issue #6 (open store only for file inputs, and ideally make it suppressible in tests) would likely resolve the allocation test. The transform-error test needs a different SQL that actually reaches the transform executor.

---

🧠 Learnings used

Learnt from: CR
Repo: strawgate/memagent PR: 0
File: coderabbit-custom-pre-merge-checks-unique-id-file-non-traceable-F7F2B60C-1728-4C9A-8889-4F2235E186CA.txt:0-0
Timestamp: 2026-04-03T03:35:42.884Z
Learning: Applies to logfwd-core/src/lifecycle.rs : PRs touching lifecycle.rs, PipelineMachine transitions, drain, checkpoint ordering, or batch sequencing must address TLA+ coverage with updated tla/PipelineMachine.tla and TLC verification output

---

Learnt from: CR
Repo: strawgate/memagent PR: 0
File: coderabbit-custom-pre-merge-checks-unique-id-file-non-traceable-F7F2B60C-1728-4C9A-8889-4F2235E186CA.txt:0-0
Timestamp: 2026-04-03T03:35:42.884Z
Learning: Applies to logfwd-io/src/**/*.rs : New async pipeline code must have proptest coverage for: arbitrary event ordering; acks out of order; drain while in-flight

---

Learnt from: CR
Repo: strawgate/memagent PR: 0
File: coderabbit-custom-pre-merge-checks-unique-id-file-non-traceable-F7F2B60C-1728-4C9A-8889-4F2235E186CA.txt:0-0
Timestamp: 2026-04-03T03:35:42.884Z
Learning: Applies to logfwd-core/src/**/*.rs : New scanner behavior must have proptest coverage for: escape sequences crossing 64-byte boundaries; fields in different orders; duplicate keys same type (first-write-wins); duplicate keys different types (two suffixed columns)

✅ Actions performed

Full review triggered.

[strawgate]

Investigated test failures:

allocation_per_row_decreases_with_scale consistently processes only 90,000/100,000 rows — even with 60s timeout (up from 30s). This is NOT a timeout issue — it's data loss: 10,000 rows are being dropped somewhere in the pipeline.

The test uses file input → null output with set_batch_timeout(10ms). With ~15MB of data (100k rows × ~150 bytes), that's about 4 batches. The last ~10k rows are consistently lost, suggesting the final batch is not being flushed on shutdown, or the checkpoint persistence changes altered how run_async drains remaining data.

Areas to investigate:

1. The run_async drain loop — does it properly flush scan_buf after channel closes?
2. Is the checkpoint store's flush() call blocking the shutdown path?
3. Does input_poll_loop properly send the final partial buffer before exiting?

The test_pipeline_transform_error_skips_batch_continues test passes locally but failed in CI — this might be a flaky race condition in the test's timing.

[strawgate]

Detailed review

Architecture: correct

Checkpoints keyed by SourceId (fingerprint), not path. Restore on startup by matching fingerprint to open files. This is the right approach — matches industry practice and our TLA+ spec.

Bug: checkpoint advances past remainder (TLA+ finding 002)

`checkpoint_data()` returns `tailed.offset` (the read position, where the fd is). `FramedInput` passes this through without subtracting `remainder.len()`. This means the checkpoint includes bytes that haven't been processed into complete lines.

On crash, the remainder is lost. On restart, the tailer seeks to the checkpoint position, which is past the partial line. Those bytes are permanently skipped.

This is exactly TLA+ finding 002 (documented in `dev-docs/research/tla-finding-002.md` on the `docs/file-tailing-audit` branch). The fix is:

```
checkpoint = read_offset - remainder.len()
```

`FramedInput.checkpoint_data()` should override the inner source's offsets by subtracting the per-source remainder length.

Bug: checkpoint advances at send time, not ack time

The checkpoint snapshot is taken in `input_poll_loop` when the batch bytes are sent through the channel. If the downstream output fails (batch dropped by transform error, output error, etc.), the checkpoint still advances. On restart, the dropped batch's data is skipped.

The checkpoint should only advance when the batch is durably delivered (acked). This requires the checkpoint to travel with the batch through the pipeline and only be persisted after ack.

Missing: per-source remainder (#797)

The shared `FramedInput` remainder bug is still present. Multiple files sharing one remainder can corrupt data across files.

Missing: crash+partial line test

The three tests cover clean shutdown and basic resume, but not the case where a crash loses partial lines in the remainder.

Recommendation

This PR has the right architecture (SourceId-keyed checkpoints, fingerprint-based restore). But the checkpoint value itself is wrong (read_offset instead of processed_offset). I'd recommend:

1. Fix `FramedInput.checkpoint_data()` to subtract remainder per source
2. Move checkpoint persistence from send-time to ack-time
3. Add a crash+partial line test

These fixes align with the TLA+ spec and `CheckpointTracker` implementation in PR #802.

[strawgate]

CodeRabbit review assessment

#	Severity	Finding	Valid?	Notes
1	🔴 Critical	Fingerprint-only restore unstable for small files	Yes	Files <1KB at checkpoint time get a different fingerprint when they grow. Restore fails silently. Need composite key or re-fingerprint on restore.
2	🔴 Critical	Test env var races	Yes	`LOGFWD_DATA_DIR` mutations race under parallel `cargo test`. Need `serial_test` or explicit `data_dir` parameter.
3	🟠 Major	Namespace checkpoint per pipeline	Yes	Multiple pipelines clobber each other's checkpoints. Need `data_dir/pipeline_name/checkpoints.json`.
4	🟠 Major	Flush throttle never resets on error	Yes	After first flush error, retries every batch, flooding logs. Need timestamp update in both branches.
5	🔵 Trivial	Docstring inconsistency	Resolved	Already fixed
6	🔵 Trivial	SAFETY comment	Resolved	Already fixed
7	🔵 Trivial	Cache persisted paths	Deferred	Nice optimization, not blocking

Additionally, our TLA+ analysis found that this PR checkpoints `read_offset` instead of `processed_offset` (read_offset - remainder.len()). See TLA+ finding 002 in PR #802. This causes data loss on crash when partial lines are in the remainder buffer.

Recommendation: Fix #3 and #4 (small changes). For #1 and the TLA+ finding, consider adopting the per-source remainder from PR #837 which provides the corrected checkpoint calculation. #2 is a test infrastructure issue — add `serial_test` or pass `data_dir` explicitly.