[reckless-sherixx]

Summary

A log reader that saves its position in a file can only save past lines whose message has already been passed on. Otherwise a restart skips lines that were read but not yet delivered.

• LogEntry gets an optional Offset: where the line ends in its source.
• The parser hands each message the Offset of the last line it consumed, as a new last argument of the message callback. That includes lines dropped over the size limit.
• While another stream's split line is still being joined, messages carry 0. A reader therefore never saves past the first part of a split line.

Breaking: the callback signature gains an argument. node-agent is the only caller, and its matching change is in a follow-up PR there.

Test plan

• go test ./... and go vet ./... pass.
• offset_test.go covers single lines, multiline messages, joined split lines across both streams, and lines over the size limit. Removing the split-line hold makes the split-line case fail.

[gemini-code-assist]

Code Review

This pull request introduces log line offset tracking within the parser and multiline collector to ensure that readers can safely resume after a crash without duplicating or skipping messages. It updates the LogEntry and Message structs to carry the offset of the last consumed line, manages offsets for partial lines from interleaved streams, and adds comprehensive unit tests. The review feedback identifies a critical edge case where interleaving split lines from different streams can lead to unsafe offset progression and duplicate or malformed log fragments upon restart. To address this, the reviewer suggests tracking a maximum unsafe offset (maxUnsafeOffset) in the parser to prevent premature offset advancement when completing or expiring interleaved split lines.

[gemini-code-assist]

There is a subtle edge case with interleaving split lines from different streams that can lead to duplicate and malformed/incomplete log fragments on restart.

The Scenario

1. Stream A Part 1 arrives at Offset 15 (Partial). partials tracks A at 15.
2. Stream B Part 1 arrives at Offset 20 (Partial). partials tracks A at 15 and B at 20.
3. Stream A Part 2 arrives at Offset 25 (Final), completing Stream A. Since len(partials) > 0 (Stream B is pending), Stream A's completed line gets Offset = 0 and is delivered.
4. Stream B never completes and eventually expires. partials.expired returns Stream B (Offset 20). Since len(expired) == 1 and len(partials) == 0, the condition len(expired) > 1 || len(partials) > 0 is false. Stream B's expired line keeps Offset = 20 and is delivered.
5. The consumer saves progress as Offset = 20 and the system crashes.
6. On restart, the reader resumes from Offset = 20, skipping Stream A's Part 1 (Offset 15) but re-reading Stream A's Part 2 (Offset 25), delivering it as a standalone, incomplete/malformed message.

Solution

To prevent this, we should track the maximum unsafe offset (e.g., by adding a maxUnsafeOffset int64 field to the Parser struct). Any subsequent line with an offset less than this maximum unsafe offset must also have its offset set to 0 to prevent unsafe progression.

					// Two streams' split lines interleave in the source, so with
					// more than one at a time no offset is safe to acknowledge.
					if len(expired) > 1 || len(partials) > 0 {
						if entry.Offset > p.maxUnsafeOffset {
							p.maxUnsafeOffset = entry.Offset
						}
						entry.Offset = 0
					} else if entry.Offset < p.maxUnsafeOffset {
						entry.Offset = 0
					} else {
						p.maxUnsafeOffset = 0
					}

[reckless-sherixx]

Good catch. The scenario is real: B's expired line acknowledged 20, a position inside A's joined line that was already passed on. A restart there re-read A's last part as a line of its own. It was a malformed duplicate, not a loss.

Fixed with a variant of your suggestion. While a split line is pending, the parser records the furthest offset it held back. When the last pending split line leaves (completed or expired), its message acknowledges the furthest offset read, since everything read by then has been passed on, and the record is cleared. Your version zeroes later lines until one passes the mark. This one moves the position forward instead.

When several split lines expire together, the last of them now acknowledges too, where before all of them carried 0.

TestPendingPartialAcknowledgesLinesPassedOnMeanwhile is your scenario: B's message carried 20 before the fix and carries 25 with it.

[gemini-code-assist]

Apply the same maximum unsafe offset tracking logic here when completing a split line while other partial lines are still pending.

					// Another stream's split line started earlier in the source and
					// is not passed on yet: acknowledging this line would skip it.
					if len(partials) > 0 {
						if entry.Offset > p.maxUnsafeOffset {
							p.maxUnsafeOffset = entry.Offset
						}
						entry.Offset = 0
					} else if entry.Offset < p.maxUnsafeOffset {
						entry.Offset = 0
					} else {
						p.maxUnsafeOffset = 0
					}

[reckless-sherixx]

Same fix, applied here: one acknowledge helper now sets the offset on both paths. See the reply on line 156.

[mayankpande88]

Looks correct to me: I could not find a case where a message acknowledges past a line that has not been passed on.

Checked:

• MultilineCollector.Add flushes the previous message before add() raises m.offset, so a new message's first line never leaks its offset into the previous one.
• Lines over the size limit raise the offset before the limit check, so they are covered.
• Split-line hold: held lines entered the collector before the line that later carries their offset, so they are in that message or an earlier one.
• Dropped lines (invalid UTF-8, decode errors, empty lines outside a message) carry no offset; the next acknowledgement covers them.
• An unflushed message at Stop() is never acknowledged, so it is re-read: at least once, as intended.

Two things:

1. A split line that never ends blocks acknowledgements for the whole source (partialLines.expired). Expiry is keyed on the latest part's arrival, so a stream that keeps producing P parts never expires, e.g. a progress bar on stdout written without a newline, which the runtime splits every 16 KiB. While it lasts, every message from the other stream carries 0, the reader's position never advances, and a restart re-reads everything since. Suggest also expiring a pending line by the age of its first part.

2. Doc nit on LogEntry.Offset: the max() logic assumes offsets only increase within a source. node-agent#407 handles truncation and rotation by encoding a generation in the high bits, but nothing here tells another caller to do that. A sentence on the field would cover it.

Neither blocks merging; (1) can be a follow-up.