From 298e696150672391c6e47e3358198e982ecd196b Mon Sep 17 00:00:00 2001 From: David McKen Date: Fri, 21 Aug 2026 04:10:20 +0000 Subject: [PATCH 1/2] fix(transport/file): write record and separator atomically Send() issued two separate Write() calls (data, then separator). With more than one decode worker calling Send() concurrently (collector's `workers` listen option), another goroutine's write can land between them, dropping the separator for that record and merging it with the next one on disk. Observed in production: a file transport running with workers=16 produced records where a fully valid new FlowMessage started at the exact byte offset a separator should have been, corrupting length-prefixed framing for any binary-format consumer built on top of it. Reducing to workers=1 eliminated the corruption immediately (confirmed clean across 1.3M+ decoded records), which pointed at a write race rather than a decode/encode bug. Fix: build the record+separator into one buffer and issue a single Write() call, which is atomic with respect to other writers on a regular file opened O_APPEND. No behavior change when a separator isn't configured. --- transport/file/transport.go | 28 ++++++++++++++++++++++------ 1 file changed, 22 insertions(+), 6 deletions(-) diff --git a/transport/file/transport.go b/transport/file/transport.go index 12626af6..14291443 100644 --- a/transport/file/transport.go +++ b/transport/file/transport.go @@ -81,21 +81,37 @@ func (d *FileDriver) Init() error { } // Send writes a formatted message and separator to the destination. +// +// The message and separator are written via a single Write() call rather +// than two. With more than one decode worker (see the collector's `workers` +// listen option), multiple goroutines can call Send() concurrently; two +// separate Write() calls for data then separator can interleave with +// another goroutine's writes landing between them, corrupting any framing +// a consumer builds on top of the separator (or on a length-prefixed binary +// format, since the separator no longer reliably follows each record). A +// single Write() call for the combined buffer is atomic with respect to +// other writers on a regular file opened O_APPEND, so concurrent Send() +// calls can no longer interleave mid-record. func (d *FileDriver) Send(key, data []byte) error { d.lock.RLock() w := d.w d.lock.RUnlock() - if len(data) > 0 { + + if d.lineSeparator == "" { + if len(data) == 0 { + return nil + } if _, err := w.Write(data); err != nil { return fmt.Errorf("write message: %w", err) } - } - if d.lineSeparator == "" { return nil } - _, err := w.Write([]byte(d.lineSeparator)) - if err != nil { - return fmt.Errorf("write separator: %w", err) + + buf := make([]byte, 0, len(data)+len(d.lineSeparator)) + buf = append(buf, data...) + buf = append(buf, d.lineSeparator...) + if _, err := w.Write(buf); err != nil { + return fmt.Errorf("write message: %w", err) } return nil } From 15605e1a984727a34a6730a0c366d4a23885697c Mon Sep 17 00:00:00 2001 From: David McKen Date: Sun, 6 Sep 2026 01:44:58 +0000 Subject: [PATCH 2/2] fix(transport/file): hold write lock for duration of Send(), not just the read Send() took RLock() only long enough to snapshot d.w, then released it before calling Write(). Init()'s SIGHUP-triggered reload goroutine takes the full Lock() to Close() the current file and open a new one (used for log rotation, e.g. `kill -1` after mv'ing the file aside). If a reopen happened between Send()'s snapshot and its Write() call, the write would fail with "file already closed" - losing whichever record was in flight at that exact moment. Observed in production: 6,498 occurrences over 8 days, clustered exactly at every 15-minute rotation boundary. Fix: hold the RLock for the duration of the write (defer RUnlock instead of releasing right after the read). Concurrent Send() calls still proceed in parallel since RLock is shared; the reopen's exclusive Lock() now blocks until in-flight writes finish, and any Send() starting after a completed reopen correctly sees the new d.w. Co-Authored-By: Claude Sonnet 5 --- transport/file/transport.go | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/transport/file/transport.go b/transport/file/transport.go index 14291443..71d02e4b 100644 --- a/transport/file/transport.go +++ b/transport/file/transport.go @@ -92,10 +92,21 @@ func (d *FileDriver) Init() error { // single Write() call for the combined buffer is atomic with respect to // other writers on a regular file opened O_APPEND, so concurrent Send() // calls can no longer interleave mid-record. +// +// The RLock is held for the duration of the write, not just the read of +// d.w: a SIGHUP-triggered reopen (see Init's reload goroutine) takes the +// write lock to Close() the current file and open a new one. If Send() +// released the read lock right after snapshotting w, a reopen could close +// that file between the snapshot and the Write() call, and the write would +// fail with "file already closed" - losing whichever records were in +// flight at the exact moment of rotation. Holding the RLock across the +// write blocks the reopen until in-flight writes finish, and readers +// (multiple Send() calls) can still run concurrently since RLock is +// shared. func (d *FileDriver) Send(key, data []byte) error { d.lock.RLock() + defer d.lock.RUnlock() w := d.w - d.lock.RUnlock() if d.lineSeparator == "" { if len(data) == 0 {