From 849951be1d1a7ee9f9302006ccb187bf5b4e36f3 Mon Sep 17 00:00:00 2001
From: Paul Buetow <paul@buetow.org>
Date: Wed, 22 Jul 2026 23:51:18 +0300
Subject: =?UTF-8?q?feat:=20DTail=20fork=20=E2=80=94=20server/client=20feat?=
 =?UTF-8?q?ure=20development?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Squashed development of the snonux/dtail fork's product code (internal/, cmd/)
since diverging from mimecast/dtail. Major areas:

- Read/output path: the former "turbo" channel-less path is now the single,
  default server-side read/output path for cat/grep/tail and MapReduce; the old
  channel-based path and its config/env toggles were removed.
- MapReduce: single aggregate implementation (server + serverless) fed directly
  by a processor pipeline, with input-exhausted finalization via the shutdown
  coordinator; high-concurrency and data-race fixes.
- Journal source reads (journal:unit.service) via journalctl, Linux-gated behind
  a journal-v1 capability.
- Auth-key fast reconnect: in-memory per-user public-key cache with TTL/max-keys,
  registered over an authenticated session (AUTHKEY), checked before
  authorized_keys.
- Interactive query reload (--interactive-query) with SESSION START/UPDATE
  generation boundaries and capability negotiation.
- Client-side deadlines: --timeout / --shutdownAfter as context deadlines;
  follow shutdown handling.
- Client logging: diagnostics-only daily log by default, opt-in payload tee via
  --log-payload.
- Numerous correctness fixes (buffer-pool double-recycle races, EOF-sentinel
  leaks, glob-expansion cap, TOCTOU in CSV parsing) with accompanying unit tests.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 internal/io/dlog/dlog.go                          | 226 ++++--
 internal/io/dlog/dlog_test.go                     |  54 ++
 internal/io/dlog/loggers/file.go                  | 122 ++-
 internal/io/dlog/loggers/file_test.go             | 183 +++++
 internal/io/dlog/loggers/fout.go                  |  68 +-
 internal/io/dlog/loggers/fout_test.go             | 199 +++++
 internal/io/dlog/loggers/stdout.go                |  86 ++-
 internal/io/dlog/loggers/stdout_test.go           | 190 +++++
 internal/io/dlog/rawlog_test.go                   |  76 ++
 internal/io/dlog/rotation.go                      |  26 +-
 internal/io/dlog/rotation_test.go                 |  61 ++
 internal/io/fs/catfile.go                         |  14 +-
 internal/io/fs/filereader.go                      |   8 +-
 internal/io/fs/permissions/permission.go          |   1 -
 internal/io/fs/permissions/permission_linuxacl.go |   1 -
 internal/io/fs/permissions/permission_test.go     |   1 -
 internal/io/fs/readfile.go                        | 238 ++----
 internal/io/fs/readfile_nozstd.go                 |  16 +
 internal/io/fs/readfile_processor.go              | 359 +++++++++
 internal/io/fs/readfile_processor_optimized.go    | 430 +++++++++++
 internal/io/fs/readfile_processor_test.go         | 869 ++++++++++++++++++++++
 internal/io/fs/readfile_zstd.go                   |  20 +
 internal/io/fs/readfilelcontext.go                | 209 ------
 internal/io/fs/rootedpath.go                      |  96 +++
 internal/io/fs/rootedpath_test.go                 |  57 ++
 internal/io/fs/tailfile.go                        |  14 +-
 internal/io/fs/validatedreadtarget.go             | 148 ++++
 internal/io/fs/validatedreadtarget_test.go        | 218 ++++++
 internal/io/journal/filter.go                     | 253 +++++++
 internal/io/journal/reader.go                     | 303 ++++++++
 internal/io/journal/reader_test.go                | 754 +++++++++++++++++++
 internal/io/journal/reader_unsupported.go         |  45 ++
 internal/io/journal/testhelper/mock.go            | 397 ++++++++++
 internal/io/journal/testhelper/mock_test.go       | 247 ++++++
 internal/io/line/line.go                          |  18 +-
 internal/io/line/processor.go                     |  22 +
 internal/io/pool/bytesbuffer.go                   |   4 +-
 internal/io/pool/scanner_pool.go                  |  85 +++
 internal/io/signal/signal.go                      |  45 ++
 39 files changed, 5623 insertions(+), 540 deletions(-)
 create mode 100644 internal/io/dlog/dlog_test.go
 create mode 100644 internal/io/dlog/loggers/file_test.go
 create mode 100644 internal/io/dlog/loggers/fout_test.go
 create mode 100644 internal/io/dlog/loggers/stdout_test.go
 create mode 100644 internal/io/dlog/rawlog_test.go
 create mode 100644 internal/io/dlog/rotation_test.go
 create mode 100644 internal/io/fs/readfile_nozstd.go
 create mode 100644 internal/io/fs/readfile_processor.go
 create mode 100644 internal/io/fs/readfile_processor_optimized.go
 create mode 100644 internal/io/fs/readfile_processor_test.go
 create mode 100644 internal/io/fs/readfile_zstd.go
 delete mode 100644 internal/io/fs/readfilelcontext.go
 create mode 100644 internal/io/fs/rootedpath.go
 create mode 100644 internal/io/fs/rootedpath_test.go
 create mode 100644 internal/io/fs/validatedreadtarget.go
 create mode 100644 internal/io/fs/validatedreadtarget_test.go
 create mode 100644 internal/io/journal/filter.go
 create mode 100644 internal/io/journal/reader.go
 create mode 100644 internal/io/journal/reader_test.go
 create mode 100644 internal/io/journal/reader_unsupported.go
 create mode 100644 internal/io/journal/testhelper/mock.go
 create mode 100644 internal/io/journal/testhelper/mock_test.go
 create mode 100644 internal/io/line/processor.go
 create mode 100644 internal/io/pool/scanner_pool.go

(limited to 'internal/io')

diff --git a/internal/io/dlog/dlog.go b/internal/io/dlog/dlog.go
index 258fb68..e28b442 100644
--- a/internal/io/dlog/dlog.go
+++ b/internal/io/dlog/dlog.go
@@ -31,36 +31,6 @@ var Common *DLog
 var mutex sync.Mutex
 var started bool
 
-// Start logger(s).
-func Start(ctx context.Context, wg *sync.WaitGroup, sourceProcess source.Source) {
-	mutex.Lock()
-	defer mutex.Unlock()
-
-	if started {
-		Common.FatalPanic("Logger already started")
-	}
-
-	Client = new(sourceProcess, source.Client)
-	Server = new(sourceProcess, source.Server)
-	Common = Client
-	if sourceProcess == source.Server {
-		Common = Server
-	}
-
-	var wg2 sync.WaitGroup
-	wg2.Add(2)
-	go Client.start(ctx, &wg2)
-	go Server.start(ctx, &wg2)
-
-	go rotation(ctx)
-	go func() {
-		wg2.Wait()
-		wg.Done()
-	}()
-
-	started = true
-}
-
 // DLog is the DTail logger.
 type DLog struct {
 	logger loggers.Logger
@@ -94,62 +64,43 @@ func new(sourceProcess, sourcePackage source.Source) *DLog {
 	}
 }
 
-func (d *DLog) start(ctx context.Context, wg *sync.WaitGroup) {
-	defer wg.Done()
-	var wg2 sync.WaitGroup
-	wg2.Add(1)
-	d.logger.Start(ctx, &wg2)
-	<-ctx.Done()
-	wg2.Wait()
-}
+// Start logger(s).
+func Start(ctx context.Context, wg *sync.WaitGroup, sourceProcess source.Source) {
+	mutex.Lock()
+	defer mutex.Unlock()
 
-func (d *DLog) log(level level, args []interface{}) string {
-	if d.maxLevel < level {
-		return ""
+	if started {
+		Common.FatalPanic("Logger already started")
 	}
-	sb := pool.BuilderBuffer.Get().(*strings.Builder)
-	defer pool.RecycleBuilderBuffer(sb)
-	now := time.Now()
 
-	switch d.sourceProcess {
-	case source.Client:
-		sb.WriteString(d.sourcePackage.String())
-		sb.WriteString(protocol.FieldDelimiter)
-		sb.WriteString(d.hostname)
-		sb.WriteString(protocol.FieldDelimiter)
-		sb.WriteString(level.String())
-	default:
-		sb.WriteString(level.String())
-		sb.WriteString(protocol.FieldDelimiter)
-		sb.WriteString(now.Format("0102-150405"))
+	Client = new(sourceProcess, source.Client)
+	Server = new(sourceProcess, source.Server)
+	Common = Client
+	if sourceProcess == source.Server {
+		Common = Server
 	}
-	sb.WriteString(protocol.FieldDelimiter)
-	d.writeArgStrings(sb, args)
 
-	message := sb.String()
-	if !config.Client.TermColorsEnable || !d.logger.SupportsColors() {
-		d.logger.Log(now, message)
-		return message
-	}
+	var wg2 sync.WaitGroup
+	wg2.Add(2)
+	go Client.start(ctx, &wg2)
+	go Server.start(ctx, &wg2)
 
-	d.logger.LogWithColors(now, message, brush.Colorfy(message))
-	return message
+	go rotation(ctx)
+	go func() {
+		wg2.Wait()
+		wg.Done()
+	}()
+
+	started = true
 }
 
-func (d *DLog) writeArgStrings(sb *strings.Builder, args []interface{}) {
-	for i, arg := range args {
-		if i > 0 {
-			sb.WriteString(protocol.FieldDelimiter)
-		}
-		switch v := arg.(type) {
-		case string:
-			sb.WriteString(v)
-		case error:
-			sb.WriteString(v.Error())
-		default:
-			sb.WriteString(fmt.Sprintf("%v", v))
-		}
-	}
+func (d *DLog) start(ctx context.Context, wg *sync.WaitGroup) {
+	defer wg.Done()
+	var wg2 sync.WaitGroup
+	wg2.Add(1)
+	d.logger.Start(ctx, &wg2)
+	<-ctx.Done()
+	wg2.Wait()
 }
 
 // FatalPanic terminates the process with a fatal error.
@@ -192,8 +143,34 @@ func (d *DLog) Debug(args ...interface{}) string {
 	return d.log(Debug, args)
 }
 
+// TraceEnabled reports whether trace-level logging is currently active.
+//
+// It performs exactly the same maxLevel comparison as Trace's internal
+// early-return (see below), letting callers on per-line hot paths gate the
+// whole trace call — the variadic []interface{} slice allocation plus the
+// interface boxing of every non-pointer argument (uint64 line counts via
+// runtime.convT64, strings via convTstring) — behind one cheap, inlinable,
+// allocation-free branch. Without this guard those args are boxed at the call
+// site before Trace even runs, so Trace's own early-return cannot save them.
+//
+// The receiver is nil-safe so call sites need no separate nil check on the
+// package-level loggers (Server/Client/Common), which stay nil until Start.
+// maxLevel is fixed at logger construction from config.Common.LogLevel, so the
+// result mirrors whatever level Trace itself would observe.
+func (d *DLog) TraceEnabled() bool {
+	return d != nil && d.maxLevel >= Trace
+}
+
 // Trace logging.
 func (d *DLog) Trace(args ...interface{}) string {
+	// Early check to avoid expensive runtime.Caller when trace is disabled
+	// This is a critical performance optimization for hot paths. Note that on
+	// per-line hot paths callers should additionally gate with TraceEnabled()
+	// so the argument boxing never happens; this check only saves runtime.Caller
+	// and the log formatting, not the caller-side boxing of args.
+	if d.maxLevel < Trace {
+		return ""
+	}
 	_, file, line, _ := runtime.Caller(1)
 	args = append(args, fmt.Sprintf("at %s:%d", file, line))
 	return d.log(Trace, args)
@@ -201,6 +178,10 @@ func (d *DLog) Trace(args ...interface{}) string {
 
 // Devel used for development purpose only logging (e.g. "print" debugging).
 func (d *DLog) Devel(args ...interface{}) string {
+	// Early check to avoid expensive runtime.Caller when devel is disabled
+	if d.maxLevel < Devel {
+		return ""
+	}
 	_, file, line, _ := runtime.Caller(1)
 	args = append(args, fmt.Sprintf("at %s:%d", file, line))
 	return d.log(Devel, args)
@@ -216,6 +197,46 @@ func (d *DLog) Raw(message string) string {
 	return message
 }
 
+// payloadFileTeer is the optional capability of a logger that can tee retrieved
+// payload into its FILE sink without also writing it to stdout. Only the default
+// fout logger implements it; stdout/none loggers have no file sink and are
+// skipped via the type assertion in RawPayloadFileTee.
+type payloadFileTeer interface {
+	RawFileOnly(now time.Time, message string)
+}
+
+// RawPayloadFileTee writes retrieved payload to the logger's FILE sink only
+// (never stdout), honoring the client's --log-payload / Client.LogPayload
+// opt-in. It is used by the serverless direct-output path, which emits payload
+// straight to stdout and therefore bypasses the fout logger's own Raw tee. When
+// the active logger has no file sink (stdout/none) or payload teeing is
+// disabled, this is a no-op. stdout output is unaffected: the caller writes the
+// same payload bytes to stdout itself, and this method only adds the file tee.
+func (d *DLog) RawPayloadFileTee(message string) {
+	if teer, ok := d.logger.(payloadFileTeer); ok {
+		teer.RawFileOnly(time.Now(), message)
+	}
+}
+
+// RawLog writes a pre-formatted message through the DIAGNOSTIC (Log) sink, so it
+// reaches both stdout and — in the default fout logger — the daily log file.
+//
+// It differs from Raw, which uses the PAYLOAD sink: with the default
+// Client.LogPayload=false, Raw is gated out of the file (bulk dcat/dgrep/dtail
+// output must not grow the log). RawLog is for audit-worthy, already-formatted
+// lines such as server-error reports, which must always be kept in the file like
+// other diagnostics rather than being treated as bulk payload. The message is
+// written verbatim (no level/hostname prefix); callers pre-format it and must
+// NOT append a trailing newline, since the Log sink appends one.
+func (d *DLog) RawLog(message string) string {
+	if !config.Client.TermColorsEnable || !d.logger.SupportsColors() {
+		d.logger.Log(time.Now(), message)
+		return message
+	}
+	d.logger.LogWithColors(time.Now(), message, brush.Colorfy(message))
+	return message
+}
+
 // Mapreduce logging.
 func (d *DLog) Mapreduce(table string, data map[string]interface{}) string {
 	args := make([]interface{}, len(data)+1)
@@ -269,3 +290,52 @@ func (d *DLog) Pause() { d.logger.Pause() }
 
 // Resume the logging.
 func (d *DLog) Resume() { d.logger.Resume() }
+
+func (d *DLog) log(level level, args []interface{}) string {
+	if d.maxLevel < level {
+		return ""
+	}
+	sb := pool.BuilderBuffer.Get().(*strings.Builder)
+	defer pool.RecycleBuilderBuffer(sb)
+	now := time.Now()
+
+	switch d.sourceProcess {
+	case source.Client:
+		sb.WriteString(d.sourcePackage.String())
+		sb.WriteString(protocol.FieldDelimiter)
+		sb.WriteString(d.hostname)
+		sb.WriteString(protocol.FieldDelimiter)
+		sb.WriteString(level.String())
+	default:
+		sb.WriteString(level.String())
+		sb.WriteString(protocol.FieldDelimiter)
+		sb.WriteString(now.Format("0102-150405"))
+	}
+	sb.WriteString(protocol.FieldDelimiter)
+	d.writeArgStrings(sb, args)
+
+	message := sb.String()
+	if !config.Client.TermColorsEnable || !d.logger.SupportsColors() {
+		d.logger.Log(now, message)
+		return message
+	}
+
+	d.logger.LogWithColors(now, message, brush.Colorfy(message))
+	return message
+}
+
+func (d *DLog) writeArgStrings(sb *strings.Builder, args []interface{}) {
+	for i, arg := range args {
+		if i > 0 {
+			sb.WriteString(protocol.FieldDelimiter)
+		}
+		switch v := arg.(type) {
+		case string:
+			sb.WriteString(v)
+		case error:
+			sb.WriteString(v.Error())
+		default:
+			sb.WriteString(fmt.Sprintf("%v", v))
+		}
+	}
+}
diff --git a/internal/io/dlog/dlog_test.go b/internal/io/dlog/dlog_test.go
new file mode 100644
index 0000000..87e834a
--- /dev/null
+++ b/internal/io/dlog/dlog_test.go
@@ -0,0 +1,54 @@
+package dlog
+
+import "testing"
+
+// TestTraceEnabled verifies that TraceEnabled mirrors Trace's internal
+// maxLevel < Trace early-return: it must report true exactly when the
+// configured level is Trace or higher (Devel/All disable trace? no — the level
+// ladder is monotonically increasing, so any level >= Trace enables trace),
+// and false for every level below Trace.
+func TestTraceEnabled(t *testing.T) {
+	tests := []struct {
+		name  string
+		level level
+		want  bool
+	}{
+		{"none", None, false},
+		{"fatal", Fatal, false},
+		{"error", Error, false},
+		{"warn", Warn, false},
+		{"info", Info, false},
+		{"default", Default, false},
+		{"verbose", Verbose, false},
+		{"debug", Debug, false},
+		{"devel", Devel, false},
+		{"trace", Trace, true},
+		{"all", All, true},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			d := &DLog{maxLevel: tc.level}
+			if got := d.TraceEnabled(); got != tc.want {
+				t.Fatalf("TraceEnabled() at level %v = %v, want %v",
+					tc.level, got, tc.want)
+			}
+			// TraceEnabled must agree with what Trace itself would do: Trace
+			// returns "" (no work) exactly when trace is disabled.
+			traceDidWork := d.maxLevel >= Trace
+			if traceDidWork != tc.want {
+				t.Fatalf("TraceEnabled() disagrees with Trace's own gate at level %v", tc.level)
+			}
+		})
+	}
+}
+
+// TestTraceEnabledNilSafe guards the footgun that call sites rely on:
+// TraceEnabled is invoked on the package-level loggers (Server/Client/Common),
+// which are nil until Start runs. A nil receiver must report false, not panic.
+func TestTraceEnabledNilSafe(t *testing.T) {
+	var d *DLog
+	if d.TraceEnabled() {
+		t.Fatal("nil *DLog.TraceEnabled() must be false")
+	}
+}
diff --git a/internal/io/dlog/loggers/file.go b/internal/io/dlog/loggers/file.go
index 8e567bc..b3b6a10 100644
--- a/internal/io/dlog/loggers/file.go
+++ b/internal/io/dlog/loggers/file.go
@@ -12,6 +12,26 @@ import (
 	"github.com/mimecast/dtail/internal/config"
 )
 
+const (
+	// fileWriterBufSize is the size of the bufio buffer wrapping the log file
+	// descriptor. A real buffer (instead of the old 1-byte writer that forced a
+	// write syscall per line) lets bulk payload — e.g. dcat/dgrep tee — batch
+	// into ~one syscall per bufferful, cutting the client receive-path syscall
+	// count and CPU by ~5x. bufio auto-flushes when full, so high-throughput
+	// output never stalls in the buffer.
+	fileWriterBufSize = 64 * 1024
+	// fileIdleFlushInterval bounds how long buffered data may sit unwritten when
+	// output goes idle (follow/interactive mode trickling a few lines). Without
+	// it, low-volume output would be stuck behind the buffer until it fills or
+	// the logger shuts down, so follow/tail would appear frozen on disk.
+	fileIdleFlushInterval = 100 * time.Millisecond
+	// fileFlushTimeout bounds how long a synchronous Flush() waits for the
+	// logger goroutine to acknowledge. It exists purely as a deadlock guard for
+	// the rare case where the goroutine is paused or already gone (e.g. Flush
+	// racing shutdown); under normal operation the ack is near-instant.
+	fileFlushTimeout = 2 * time.Second
+)
+
 type fileMessageBuf struct {
 	now     time.Time
 	message string
@@ -23,7 +43,13 @@ type file struct {
 	pauseCh      chan struct{}
 	resumeCh     chan struct{}
 	rotateCh     chan struct{}
-	flushCh      chan struct{}
+	// flushCh carries a per-call reply channel so Flush() can block until the
+	// logger goroutine has actually drained the buffer channel and flushed the
+	// bufio writer to disk. This makes Flush() synchronous, which the crash path
+	// (dlog.FatalPanic -> Flush -> panic) relies on: an async signal could let
+	// the process unwind before the goroutine drains, dropping up to one buffer
+	// (64KB) of Fatal diagnostics.
+	flushCh      chan chan struct{}
 	fd           *os.File
 	writer       *bufio.Writer
 	mutex        sync.Mutex
@@ -32,13 +58,20 @@ type file struct {
 	strategy     Strategy
 }
 
+var _ Logger = (*file)(nil)
+
 func newFile(strategy Strategy) *file {
+	// Pause/Resume/Rotate use capacity-1, non-blocking coalescing sends so
+	// callers never block on the logger goroutine (repeated signals collapse
+	// into one pending notification). flushCh is unbuffered and carries a reply
+	// channel because Flush() is synchronous: it must wait for the goroutine to
+	// drain and write before returning.
 	return &file{
 		bufferCh: make(chan *fileMessageBuf, runtime.NumCPU()*100),
-		pauseCh:  make(chan struct{}),
-		resumeCh: make(chan struct{}),
-		rotateCh: make(chan struct{}),
-		flushCh:  make(chan struct{}),
+		pauseCh:  make(chan struct{}, 1),
+		resumeCh: make(chan struct{}, 1),
+		rotateCh: make(chan struct{}, 1),
+		flushCh:  make(chan chan struct{}),
 		strategy: strategy,
 	}
 }
@@ -67,17 +100,43 @@ func (f *file) Start(ctx context.Context, wg *sync.WaitGroup) {
 
 	go func() {
 		defer wg.Done()
+		// Idle-flush ticker: with a real (64KB) buffer, low-volume output
+		// (follow/interactive) would otherwise sit in the buffer until it
+		// fills. The ticker flushes any partial buffer promptly so follow/tail
+		// output reaches disk within fileIdleFlushInterval. flush() is cheap
+		// when nothing is buffered.
+		ticker := time.NewTicker(fileIdleFlushInterval)
+		defer ticker.Stop()
 		for {
 			select {
 			case m := <-f.bufferCh:
 				f.write(m)
+			case <-ticker.C:
+				f.flush()
 			case <-f.pauseCh:
+				// Flush before pausing so all output produced so far is on
+				// disk before the caller (e.g. an interactive prompt) writes
+				// directly to the terminal/file; preserves ordering.
+				f.flush()
 				pause(ctx)
-			case <-f.flushCh:
+			case done := <-f.flushCh:
+				// Synchronous flush: drain + write, then acknowledge so the
+				// blocked Flush() caller can proceed (used by FatalPanic).
 				f.flush()
+				close(done)
+			case <-f.rotateCh:
+				// Force re-opening the outfile on the next write.
+				// Drained here (not only from write()) so that Rotate()
+				// makes progress even when no log messages arrive.
+				f.lastFileName = ""
 			case <-ctx.Done():
 				f.flush()
-				f.fd.Close()
+				// f.fd is only populated after the first getWriter() call;
+				// guard against a nil pointer when the logger is shut down
+				// before anything has been written.
+				if f.fd != nil {
+					f.fd.Close()
+				}
 				return
 			}
 		}
@@ -100,21 +159,43 @@ func (f *file) RawWithColors(now time.Time, message, coloredMessage string) {
 	panic("Colors not supported in file logger")
 }
 
-func (f *file) Pause()  { f.pauseCh <- struct{}{} }
-func (f *file) Resume() { f.resumeCh <- struct{}{} }
-func (f *file) Flush()  { f.flushCh <- struct{}{} }
+// signal performs a non-blocking, coalescing send on a capacity-1 control
+// channel. If a signal is already pending the new one is dropped, which is
+// the desired behaviour for idempotent operations such as Pause/Rotate/Flush.
+func signal(ch chan struct{}) {
+	select {
+	case ch <- struct{}{}:
+	default:
+	}
+}
 
-func (f *file) Rotate()            { f.rotateCh <- struct{}{} }
-func (*file) SupportsColors() bool { return false }
+func (f *file) Pause()  { signal(f.pauseCh) }
+func (f *file) Resume() { signal(f.resumeCh) }
+func (f *file) Rotate() { signal(f.rotateCh) }
 
-func (f *file) write(m *fileMessageBuf) {
+// Flush synchronously drains any queued messages and writes the bufio buffer to
+// disk, blocking until the logger goroutine acknowledges. The crash path
+// (dlog.FatalPanic) depends on this: with an async signal the process could
+// panic and unwind before the goroutine drained, losing buffered diagnostics.
+// A bounded timeout guards against a deadlock when the goroutine is paused or
+// has already exited (Flush racing shutdown), in which case the ctx.Done path
+// has already flushed or will flush.
+func (f *file) Flush() {
+	done := make(chan struct{})
 	select {
-	case <-f.rotateCh:
-		// Force re-opening the outfile next time in getWriter.
-		f.lastFileName = ""
-	default:
+	case f.flushCh <- done:
+	case <-time.After(fileFlushTimeout):
+		return
+	}
+	select {
+	case <-done:
+	case <-time.After(fileFlushTimeout):
 	}
+}
+
+func (*file) SupportsColors() bool { return false }
 
+func (f *file) write(m *fileMessageBuf) {
 	var writer *bufio.Writer
 	if f.strategy.Rotation == DailyRotation {
 		writer = f.getWriter(m.now.Format("20060102"))
@@ -150,9 +231,12 @@ func (f *file) getWriter(name string) *bufio.Writer {
 		f.writer.Flush()
 		f.fd.Close()
 	}
-	// Set new writer.
+	// Set new writer. Use a real buffer (fileWriterBufSize) so bulk payload
+	// batches into few write syscalls instead of one-or-two per line. The
+	// logger goroutine's idle ticker and the ctx.Done/flush/pause paths keep
+	// low-volume and shutdown output from being stuck in the buffer.
 	f.fd = newFd
-	f.writer = bufio.NewWriterSize(f.fd, 1)
+	f.writer = bufio.NewWriterSize(f.fd, fileWriterBufSize)
 	f.lastFileName = name
 
 	return f.writer
diff --git a/internal/io/dlog/loggers/file_test.go b/internal/io/dlog/loggers/file_test.go
new file mode 100644
index 0000000..0436327
--- /dev/null
+++ b/internal/io/dlog/loggers/file_test.go
@@ -0,0 +1,183 @@
+package loggers
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/mimecast/dtail/internal/config"
+)
+
+// withTempLogDir points config.Common.LogDir at a fresh temp dir for the
+// duration of a test and restores the previous config afterwards. The file
+// logger resolves its output path from config.Common.LogDir at write time.
+func withTempLogDir(t *testing.T) string {
+	t.Helper()
+	dir := t.TempDir()
+	prev := config.Common
+	config.Common = &config.CommonConfig{LogDir: dir}
+	t.Cleanup(func() { config.Common = prev })
+	return dir
+}
+
+// startFileLogger starts f and returns a stop func that cancels the context and
+// JOINS the logger goroutine (wg.Wait). Tests must defer stop() so the goroutine
+// has fully exited before returning: withTempLogDir's t.Cleanup restores the
+// global config.Common, and a still-running goroutine reading config.Common.LogDir
+// would otherwise race that restore. For the same reason none of these tests may
+// call t.Parallel — they mutate the process-global config.Common.
+func startFileLogger(t *testing.T, f *file) func() {
+	t.Helper()
+	ctx, cancel := context.WithCancel(context.Background())
+	var wg sync.WaitGroup
+	wg.Add(1)
+	f.Start(ctx, &wg)
+	return func() {
+		cancel()
+		wg.Wait()
+	}
+}
+
+func readLogFile(t *testing.T, dir, base string) string {
+	t.Helper()
+	data, err := os.ReadFile(filepath.Join(dir, base+".log"))
+	if err != nil {
+		if os.IsNotExist(err) {
+			return ""
+		}
+		t.Fatalf("reading log file: %v", err)
+	}
+	return string(data)
+}
+
+// TestFileLoggerNothingLostOnClose verifies every logged line reaches disk when
+// the context is cancelled (clean shutdown): the goroutine drains the buffer
+// channel and flushes the 64KB writer before closing the fd.
+func TestFileLoggerNothingLostOnClose(t *testing.T) {
+	dir := withTempLogDir(t)
+	base := "close-test"
+	f := newFile(Strategy{Rotation: SignalRotation, FileBase: base})
+	stop := startFileLogger(t, f)
+
+	const n = 500
+	var want strings.Builder
+	for i := 0; i < n; i++ {
+		line := "line-" + strconv.Itoa(i)
+		f.Log(time.Now(), line)
+		want.WriteString(line + "\n")
+	}
+
+	// stop() cancels the context and joins the goroutine, which flushes and
+	// closes the fd on the way out — so all output must be on disk afterwards.
+	stop()
+
+	if got := readLogFile(t, dir, base); got != want.String() {
+		t.Fatalf("lost output on close: got %d bytes, want %d bytes",
+			len(got), want.Len())
+	}
+}
+
+// TestFileLoggerIdleFlush verifies a single low-volume line (follow/tail style)
+// is not stuck behind the 64KB buffer: the idle ticker flushes it to disk
+// promptly without any explicit Flush or shutdown. The logger goroutine is
+// joined via stop() before returning so it cannot outlive config.Common.
+func TestFileLoggerIdleFlush(t *testing.T) {
+	dir := withTempLogDir(t)
+	base := "idle-test"
+	f := newFile(Strategy{Rotation: SignalRotation, FileBase: base})
+	stop := startFileLogger(t, f)
+	defer stop()
+
+	f.Log(time.Now(), "follow-line")
+
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		if strings.Contains(readLogFile(t, dir, base), "follow-line") {
+			return
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	t.Fatal("follow-style line stuck behind buffer; idle flush did not emit it")
+}
+
+// TestFileLoggerExplicitFlush verifies Flush() is SYNCHRONOUS: once it returns,
+// the buffered data is already on disk (no polling needed). This is the property
+// dlog.FatalPanic relies on to not drop Fatal diagnostics before panicking.
+func TestFileLoggerExplicitFlush(t *testing.T) {
+	dir := withTempLogDir(t)
+	base := "flush-test"
+	f := newFile(Strategy{Rotation: SignalRotation, FileBase: base})
+	stop := startFileLogger(t, f)
+	defer stop()
+
+	f.Log(time.Now(), "flush-me")
+	f.Flush()
+
+	if got := readLogFile(t, dir, base); !strings.Contains(got, "flush-me") {
+		t.Fatalf("synchronous Flush() did not persist data before returning; got %q", got)
+	}
+}
+
+// TestFileLoggerRotateDoesNotBlockWithoutWrites verifies that Rotate() does
+// not deadlock when no log messages have been produced. Previously rotateCh
+// was unbuffered and only drained opportunistically from write(), so a SIGHUP
+// before any Log() call would block the caller forever.
+func TestFileLoggerRotateDoesNotBlockWithoutWrites(t *testing.T) {
+	f := newFile(Strategy{Rotation: SignalRotation, FileBase: "unit-test"})
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+	f.Start(ctx, &wg)
+
+	done := make(chan struct{})
+	go func() {
+		f.Rotate()
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(500 * time.Millisecond):
+		t.Fatal("Rotate() blocked without any writes; expected prompt return")
+	}
+
+	cancel()
+	wg.Wait()
+}
+
+// TestFileLoggerCancelBeforeFirstWriteDoesNotPanic verifies that cancelling
+// the context before any write has happened does not panic. Previously the
+// goroutine called f.fd.Close() unconditionally, but f.fd is only populated
+// by the first getWriter() call, so a ctx cancel with no prior writes
+// panicked on a nil pointer.
+func TestFileLoggerCancelBeforeFirstWriteDoesNotPanic(t *testing.T) {
+	f := newFile(Strategy{Rotation: SignalRotation, FileBase: "unit-test"})
+
+	ctx, cancel := context.WithCancel(context.Background())
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+	f.Start(ctx, &wg)
+
+	cancel()
+
+	doneCh := make(chan struct{})
+	go func() {
+		wg.Wait()
+		close(doneCh)
+	}()
+
+	select {
+	case <-doneCh:
+	case <-time.After(1 * time.Second):
+		t.Fatal("file logger goroutine did not exit after ctx cancel")
+	}
+}
diff --git a/internal/io/dlog/loggers/fout.go b/internal/io/dlog/loggers/fout.go
index 6888d40..0d3fde5 100644
--- a/internal/io/dlog/loggers/fout.go
+++ b/internal/io/dlog/loggers/fout.go
@@ -4,16 +4,47 @@ import (
 	"context"
 	"sync"
 	"time"
+
+	"github.com/mimecast/dtail/internal/config"
 )
 
+// fout logs to both a file and stdout. It is the default client logger.
+//
+// The two things a client emits are deliberately split at this seam:
+//   - Diagnostics (connection INFO/WARN/ERROR/etc.) arrive via Log/LogWithColors
+//     and are ALWAYS written to both stdout and the file — they are the small,
+//     useful audit trail the daily log file is meant to keep.
+//   - Retrieved payload (the bulk dcat/dgrep/dtail output) arrives via
+//     Raw/RawWithColors. It always reaches stdout/terminal, but it is teed to
+//     the file only when logPayload is set (opt-in via --log-payload /
+//     Client.LogPayload). By default the file receives no payload, so a bulk
+//     dcat no longer silently grows the daily log file by the full payload size.
 type fout struct {
-	file   *file
-	stdout *stdout
+	file       Logger
+	stdout     Logger
+	logPayload bool
 }
 
-// Logs to both, a file and stdout
+// newFout builds the default client logger. Whether retrieved payload is teed
+// to the file is decided once at construction from the client config.
 func newFout(strategy Strategy) *fout {
-	return &fout{file: newFile(strategy), stdout: newStdout()}
+	return newFoutWithSinks(newFile(strategy), newStdout(), clientLogPayloadEnabled())
+}
+
+// newFoutWithSinks builds a fout over injectable sinks and an explicit payload
+// switch. Production uses newFout (concrete file+stdout, config-driven switch);
+// tests inject fakes to assert that diagnostics always reach the file while
+// payload reaches it only when opted in.
+func newFoutWithSinks(file, stdout Logger, logPayload bool) *fout {
+	return &fout{file: file, stdout: stdout, logPayload: logPayload}
+}
+
+// clientLogPayloadEnabled reports whether the client has opted in to teeing the
+// full retrieved payload into the daily log file. Default (false) keeps only
+// diagnostics in the file. config.Client is nil-guarded because a logger can be
+// constructed in early/unit contexts before config.Setup has populated it.
+func clientLogPayloadEnabled() bool {
+	return config.Client != nil && config.Client.LogPayload
 }
 
 func (f *fout) Start(ctx context.Context, wg *sync.WaitGroup) {
@@ -35,17 +66,42 @@ func (f *fout) Log(now time.Time, message string) {
 
 func (f *fout) LogWithColors(now time.Time, message, coloredMessage string) {
 	f.stdout.LogWithColors(now, "", coloredMessage)
+	// The file logger does not support colors, so write the plain message via
+	// Log (its LogWithColors would route to RawWithColors, which panics).
 	f.file.Log(now, message)
 }
 
+// Raw writes retrieved payload. It always reaches stdout/terminal; it is teed
+// to the file sink only when the client opted in via --log-payload /
+// Client.LogPayload. By default the file is left payload-free.
 func (f *fout) Raw(now time.Time, message string) {
 	f.stdout.Raw(now, message)
-	f.file.Raw(now, message)
+	if f.logPayload {
+		f.file.Raw(now, message)
+	}
 }
 
 func (f *fout) RawWithColors(now time.Time, message, coloredMessage string) {
 	f.stdout.RawWithColors(now, "", coloredMessage)
-	f.file.Raw(now, message)
+	// Same opt-in gate as Raw; the file gets the plain (uncolored) payload.
+	if f.logPayload {
+		f.file.Raw(now, message)
+	}
+}
+
+// RawFileOnly tees retrieved payload into the daily log FILE sink only, never to
+// stdout, honoring the same --log-payload / Client.LogPayload opt-in as Raw.
+//
+// It exists for the serverless direct-output path: that path writes
+// payload straight to its own stdout sink and bypasses Raw entirely, so without
+// this hook --log-payload would silently no longer tee payload to the file in
+// serverless mode. The caller (the serverless output writer) already emits the
+// payload bytes to stdout itself, so this method deliberately writes ONLY to the
+// file to keep stdout byte-identical whether or not --log-payload is set.
+func (f *fout) RawFileOnly(now time.Time, message string) {
+	if f.logPayload {
+		f.file.Raw(now, message)
+	}
 }
 
 func (f *fout) Flush()  { f.stdout.Flush(); f.file.Flush() }
diff --git a/internal/io/dlog/loggers/fout_test.go b/internal/io/dlog/loggers/fout_test.go
new file mode 100644
index 0000000..0788f01
--- /dev/null
+++ b/internal/io/dlog/loggers/fout_test.go
@@ -0,0 +1,199 @@
+package loggers
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/mimecast/dtail/internal/config"
+)
+
+// recordingSink is an injectable Logger that records which messages reached it
+// via the diagnostic path (Log) versus the payload path (Raw). It lets the fout
+// routing tests assert exactly which sink gets diagnostics and which gets
+// payload without touching real files or stdout.
+type recordingSink struct {
+	mutex sync.Mutex
+	logs  []string
+	raws  []string
+}
+
+func (r *recordingSink) Log(now time.Time, message string) {
+	r.mutex.Lock()
+	defer r.mutex.Unlock()
+	r.logs = append(r.logs, message)
+}
+
+func (r *recordingSink) LogWithColors(now time.Time, message, colored string) {
+	r.Log(now, message)
+}
+
+func (r *recordingSink) Raw(now time.Time, message string) {
+	r.mutex.Lock()
+	defer r.mutex.Unlock()
+	r.raws = append(r.raws, message)
+}
+
+func (r *recordingSink) RawWithColors(now time.Time, message, colored string) {
+	r.Raw(now, message)
+}
+
+func (r *recordingSink) Start(ctx context.Context, wg *sync.WaitGroup) { wg.Done() }
+func (r *recordingSink) Flush()                                        {}
+func (r *recordingSink) Pause()                                        {}
+func (r *recordingSink) Resume()                                       {}
+func (r *recordingSink) Rotate()                                       {}
+func (r *recordingSink) SupportsColors() bool                          { return false }
+
+func (r *recordingSink) logCount() int {
+	r.mutex.Lock()
+	defer r.mutex.Unlock()
+	return len(r.logs)
+}
+
+func (r *recordingSink) rawCount() int {
+	r.mutex.Lock()
+	defer r.mutex.Unlock()
+	return len(r.raws)
+}
+
+var _ Logger = (*recordingSink)(nil)
+
+// TestFoutDefaultKeepsPayloadOutOfFile proves the footgun fix: with LogPayload
+// disabled (the default), diagnostics reach the file sink but retrieved payload
+// does not — while stdout still receives BOTH, so the terminal output is
+// unchanged.
+func TestFoutDefaultKeepsPayloadOutOfFile(t *testing.T) {
+	file := &recordingSink{}
+	stdout := &recordingSink{}
+	f := newFoutWithSinks(file, stdout, false)
+
+	now := time.Now()
+	f.Log(now, "diagnostic-line")  // connection INFO/WARN/ERROR audit line
+	f.Raw(now, "payload-line-1\n") // bulk dcat/dgrep/dtail output
+	f.Raw(now, "payload-line-2\n")
+
+	// File: exactly the diagnostic, no payload.
+	if got := file.logCount(); got != 1 {
+		t.Fatalf("file diagnostics: got %d, want 1", got)
+	}
+	if got := file.rawCount(); got != 0 {
+		t.Fatalf("file payload leaked: got %d raws, want 0 (default must be diagnostics-only)", got)
+	}
+	// Stdout: diagnostic + both payload lines (terminal output unchanged).
+	if got := stdout.logCount(); got != 1 {
+		t.Fatalf("stdout diagnostics: got %d, want 1", got)
+	}
+	if got := stdout.rawCount(); got != 2 {
+		t.Fatalf("stdout payload: got %d, want 2", got)
+	}
+}
+
+// TestFoutOptInTeesPayloadToFile proves --log-payload / Client.LogPayload
+// restores the legacy behaviour: payload is teed to the file too, and
+// diagnostics still land in the file.
+func TestFoutOptInTeesPayloadToFile(t *testing.T) {
+	file := &recordingSink{}
+	stdout := &recordingSink{}
+	f := newFoutWithSinks(file, stdout, true)
+
+	now := time.Now()
+	f.Log(now, "diagnostic-line")
+	f.Raw(now, "payload-line-1\n")
+	f.Raw(now, "payload-line-2\n")
+
+	if got := file.logCount(); got != 1 {
+		t.Fatalf("file diagnostics: got %d, want 1", got)
+	}
+	if got := file.rawCount(); got != 2 {
+		t.Fatalf("file payload with opt-in: got %d, want 2", got)
+	}
+	if got := stdout.rawCount(); got != 2 {
+		t.Fatalf("stdout payload: got %d, want 2", got)
+	}
+}
+
+// TestFoutServerErrorDiagnosticReachesFileByDefault is the end-to-end guard for
+// the ReportServerError footgun: a server-error audit line, sent via the
+// diagnostic (Log) path over a REAL on-disk file sink with LogPayload=false,
+// must actually land in the daily log file — while bulk payload (Raw) must not.
+// This exercises the real file-write path the regression would have skipped.
+func TestFoutServerErrorDiagnosticReachesFileByDefault(t *testing.T) {
+	tmp := t.TempDir()
+	prevCommon := config.Common
+	config.Common = &config.CommonConfig{LogDir: tmp}
+	t.Cleanup(func() { config.Common = prevCommon })
+
+	// SignalRotation with a fixed FileBase gives a deterministic file name.
+	fileSink := newFile(Strategy{Rotation: SignalRotation, FileBase: "servererr"})
+	stdout := &recordingSink{}
+	f := newFoutWithSinks(fileSink, stdout, false)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	var wg sync.WaitGroup
+	wg.Add(1)
+	f.Start(ctx, &wg)
+
+	now := time.Now()
+	serverError := "SERVER|srv1|ERROR|journal file targets require server capability journal-v1"
+	f.Log(now, serverError) // diagnostic / audit line (ReportServerError path)
+	f.Raw(now, "payload\n") // bulk payload, must stay out of the file
+
+	f.Flush()
+	cancel()
+	wg.Wait()
+
+	content, err := os.ReadFile(filepath.Join(tmp, "servererr.log"))
+	if err != nil {
+		t.Fatalf("reading log file: %v", err)
+	}
+	if !strings.Contains(string(content), serverError) {
+		t.Fatalf("server-error diagnostic missing from default log file:\n%s", content)
+	}
+	if strings.Contains(string(content), "payload") {
+		t.Fatalf("payload leaked into the default log file:\n%s", content)
+	}
+	// Stdout still receives both (terminal output unchanged).
+	if stdout.logCount() != 1 || stdout.rawCount() != 1 {
+		t.Fatalf("stdout must receive both diagnostic and payload; logs=%d raws=%d",
+			stdout.logCount(), stdout.rawCount())
+	}
+}
+
+// TestFoutWithColorsRouting mirrors the two tests above for the colored paths
+// (LogWithColors always to file, RawWithColors gated by the opt-in), because the
+// client uses the *WithColors variants when terminal colors are enabled.
+func TestFoutWithColorsRouting(t *testing.T) {
+	now := time.Now()
+
+	t.Run("default", func(t *testing.T) {
+		file := &recordingSink{}
+		stdout := &recordingSink{}
+		f := newFoutWithSinks(file, stdout, false)
+		f.LogWithColors(now, "diag", "\x1b[1mdiag\x1b[0m")
+		f.RawWithColors(now, "payload\n", "\x1b[1mpayload\x1b[0m\n")
+		if got := file.logCount(); got != 1 {
+			t.Fatalf("file diagnostics: got %d, want 1", got)
+		}
+		if got := file.rawCount(); got != 0 {
+			t.Fatalf("file payload leaked: got %d, want 0", got)
+		}
+		if got := stdout.rawCount(); got != 1 {
+			t.Fatalf("stdout payload: got %d, want 1", got)
+		}
+	})
+
+	t.Run("optin", func(t *testing.T) {
+		file := &recordingSink{}
+		stdout := &recordingSink{}
+		f := newFoutWithSinks(file, stdout, true)
+		f.RawWithColors(now, "payload\n", "\x1b[1mpayload\x1b[0m\n")
+		if got := file.rawCount(); got != 1 {
+			t.Fatalf("file payload with opt-in: got %d, want 1", got)
+		}
+	})
+}
diff --git a/internal/io/dlog/loggers/stdout.go b/internal/io/dlog/loggers/stdout.go
index b024243..7067305 100644
--- a/internal/io/dlog/loggers/stdout.go
+++ b/internal/io/dlog/loggers/stdout.go
@@ -1,27 +1,76 @@
 package loggers
 
 import (
+	"bufio"
 	"context"
 	"fmt"
+	"io"
+	"os"
 	"sync"
 	"time"
 )
 
+const (
+	// stdoutWriterBufSize is the size of the bufio buffer wrapping os.Stdout.
+	// The old path did one fmt.Println (one write syscall) per received line;
+	// buffering lets bulk payload batch into ~one syscall per bufferful. bufio
+	// auto-flushes when full so high-throughput output never stalls.
+	stdoutWriterBufSize = 64 * 1024
+	// stdoutIdleFlushInterval bounds how long buffered output may sit unwritten
+	// when output goes idle (follow/interactive trickling a few lines). Without
+	// it, low-volume output would be stuck behind the buffer, so follow/tail
+	// would appear frozen on the terminal.
+	stdoutIdleFlushInterval = 100 * time.Millisecond
+)
+
 type stdout struct {
 	pauseCh  chan struct{}
 	resumeCh chan struct{}
+	writer   *bufio.Writer
 	mutex    sync.Mutex
 }
 
+var _ Logger = (*stdout)(nil)
+
 func newStdout() *stdout {
+	return newStdoutWriter(os.Stdout)
+}
+
+// newStdoutWriter builds a stdout logger over an arbitrary sink. Production
+// uses os.Stdout; tests inject a counting writer to assert that buffering
+// batches many lines into few underlying writes. The bufio writer is created
+// eagerly so the logger is usable even when Start() is never called (e.g. in
+// isolated unit tests); idle/shutdown flushing is only driven once Start()
+// spawns the flush goroutine.
+func newStdoutWriter(w io.Writer) *stdout {
 	return &stdout{
 		pauseCh:  make(chan struct{}),
 		resumeCh: make(chan struct{}),
+		writer:   bufio.NewWriterSize(w, stdoutWriterBufSize),
 	}
 }
 
 func (s *stdout) Start(ctx context.Context, wg *sync.WaitGroup) {
-	wg.Done()
+	// Background flusher: with a real buffer, low-volume (follow/interactive)
+	// output would otherwise sit unwritten until the buffer fills. The ticker
+	// flushes any partial buffer promptly, and ctx.Done triggers a final flush
+	// so no buffered output is lost on clean shutdown. wg.Done is deferred to
+	// the goroutine so callers (ClientRuntime.Stop -> wg.Wait) block until the
+	// final flush has happened.
+	go func() {
+		defer wg.Done()
+		ticker := time.NewTicker(stdoutIdleFlushInterval)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ticker.C:
+				s.Flush()
+			case <-ctx.Done():
+				s.Flush()
+				return
+			}
+		}
+	}()
 }
 
 func (s *stdout) Log(now time.Time, message string) {
@@ -42,27 +91,48 @@ func (s *stdout) RawWithColors(now time.Time, message, coloredMessage string) {
 
 func (s *stdout) log(message string, nl bool) {
 	s.mutex.Lock()
-	defer s.mutex.Unlock()
-
 	select {
 	case <-s.pauseCh:
-		// Pause until resumed.
+		// Wait for Resume without holding the mutex: the prompt path calls
+		// dlog after the user answers while Pause is still active; holding the
+		// mutex here would deadlock (Info blocks on Lock, Resume never runs).
+		s.mutex.Unlock()
 		<-s.resumeCh
+		s.mutex.Lock()
 	default:
 	}
+	defer s.mutex.Unlock()
 
+	// Buffered writes: fmt.Fprint(ln) into the bufio.Writer batches many lines
+	// into one write syscall. Errors are intentionally ignored — a logger that
+	// cannot write to stdout has nowhere to report the failure.
 	if nl {
-		fmt.Println(message)
+		_, _ = fmt.Fprintln(s.writer, message)
 		return
 	}
-	fmt.Print(message)
+	_, _ = fmt.Fprint(s.writer, message)
+}
+
+func (s *stdout) Pause() {
+	// Flush before pausing so all output produced so far is visible before the
+	// caller (interactive prompt / stats interrupt) writes directly to stdout,
+	// preserving the ordering the unbuffered path used to give for free. The
+	// pauseCh handshake below is unchanged so the pause semantics (and the
+	// deadlock guarantees exercised by the unit tests) are preserved.
+	s.mutex.Lock()
+	_ = s.writer.Flush()
+	s.mutex.Unlock()
+	s.pauseCh <- struct{}{}
 }
 
-func (s *stdout) Pause()  { s.pauseCh <- struct{}{} }
 func (s *stdout) Resume() { s.resumeCh <- struct{}{} }
 
 func (s *stdout) Flush() {
-	// This is empty because it isn't doing anything but has to satisfy the interface.
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+	// bufio.Flush is a no-op when nothing is buffered, so calling this on every
+	// idle tick is cheap.
+	_ = s.writer.Flush()
 }
 
 func (s *stdout) Rotate() {
diff --git a/internal/io/dlog/loggers/stdout_test.go b/internal/io/dlog/loggers/stdout_test.go
new file mode 100644
index 0000000..af8c9e5
--- /dev/null
+++ b/internal/io/dlog/loggers/stdout_test.go
@@ -0,0 +1,190 @@
+package loggers
+
+import (
+	"bytes"
+	"context"
+	"strconv"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+)
+
+// countingWriter records how many times Write is called and accumulates all
+// bytes, so tests can assert that buffering batches many logical lines into a
+// small number of underlying writes while preserving content and order.
+type countingWriter struct {
+	mutex  sync.Mutex
+	buf    bytes.Buffer
+	writes int
+}
+
+func (c *countingWriter) Write(p []byte) (int, error) {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+	c.writes++
+	return c.buf.Write(p)
+}
+
+func (c *countingWriter) String() string {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+	return c.buf.String()
+}
+
+func (c *countingWriter) Writes() int {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+	return c.writes
+}
+
+// TestStdoutBuffersAndPreservesOrder proves the buffered stdout path batches
+// many lines into far fewer underlying writes than one-per-line, and that a
+// Flush emits the exact content in order (nothing dropped or reordered).
+func TestStdoutBuffersAndPreservesOrder(t *testing.T) {
+	cw := &countingWriter{}
+	s := newStdoutWriter(cw)
+
+	const n = 1000
+	var want strings.Builder
+	for i := 0; i < n; i++ {
+		line := "line-" + strconv.Itoa(i)
+		s.Raw(time.Now(), line+"\n")
+		want.WriteString(line + "\n")
+	}
+
+	// Before flush the small lines must still be batched in the bufio buffer:
+	// with per-line writes this would already be n writes.
+	if got := cw.Writes(); got >= n {
+		t.Fatalf("expected buffering to batch writes, got %d writes for %d lines", got, n)
+	}
+
+	s.Flush()
+
+	if got := cw.String(); got != want.String() {
+		t.Fatalf("content mismatch after flush:\n got %q\nwant %q", got, want.String())
+	}
+	// 1000 short lines fit in a handful of 64KB flushes, definitely far below n.
+	if got := cw.Writes(); got > n/10 {
+		t.Fatalf("expected far fewer than %d writes, got %d", n/10, got)
+	}
+}
+
+// TestStdoutFlushOnPause proves output produced before Pause() is flushed to
+// the sink before Pause returns, so an interactive prompt writing directly to
+// the terminal never appears ahead of already-logged output.
+func TestStdoutFlushOnPause(t *testing.T) {
+	cw := &countingWriter{}
+	s := newStdoutWriter(cw)
+
+	s.Raw(time.Now(), "before-pause\n")
+
+	// Pause blocks on the pauseCh handshake until a concurrent log() consumes
+	// the token (the existing pause semantics). Drive a stream of log() calls
+	// so one is guaranteed to pick up the token and let Pause() return,
+	// mirroring the real prompt flow where logging goroutines are active.
+	paused := make(chan struct{})
+	go func() {
+		s.Pause()
+		close(paused)
+	}()
+	go func() {
+		for {
+			select {
+			case <-paused:
+				return
+			default:
+				s.Log(time.Now(), "consumes-pause")
+				time.Sleep(time.Millisecond)
+			}
+		}
+	}()
+	<-paused
+
+	// Pause() flushes before the handshake, so the pre-pause line must already
+	// be in the sink now regardless of the buffer.
+	if got := cw.String(); !strings.Contains(got, "before-pause") {
+		t.Fatalf("expected buffered output flushed on Pause, got %q", got)
+	}
+	s.Resume()
+}
+
+// TestStdoutIdleFlush proves that a single low-volume line (follow/tail style)
+// is not stuck behind the buffer: the Start() idle ticker flushes it promptly
+// without any explicit Flush call.
+func TestStdoutIdleFlush(t *testing.T) {
+	cw := &countingWriter{}
+	s := newStdoutWriter(cw)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	var wg sync.WaitGroup
+	wg.Add(1)
+	s.Start(ctx, &wg)
+
+	s.Raw(time.Now(), "follow-line\n")
+
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		if strings.Contains(cw.String(), "follow-line") {
+			cancel()
+			wg.Wait()
+			return
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	cancel()
+	wg.Wait()
+	t.Fatal("follow-style line stuck behind buffer; idle flush did not emit it")
+}
+
+// TestStdoutFinalFlushOnClose proves no buffered output is lost on clean
+// shutdown: data logged just before ctx cancel is flushed before the Start
+// goroutine (and thus wg.Wait) returns.
+func TestStdoutFinalFlushOnClose(t *testing.T) {
+	cw := &countingWriter{}
+	s := newStdoutWriter(cw)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	var wg sync.WaitGroup
+	wg.Add(1)
+	s.Start(ctx, &wg)
+
+	s.Raw(time.Now(), "last-line-before-exit\n")
+	cancel()
+	wg.Wait()
+
+	if got := cw.String(); !strings.Contains(got, "last-line-before-exit") {
+		t.Fatalf("buffered output lost on shutdown, got %q", got)
+	}
+}
+
+// Regression: during an interactive prompt, dlog.Common.Pause() unblocks when some
+// goroutine hits stdout.log(); that goroutine must not hold the stdout mutex while
+// waiting on resume, or dlog.Client.Info from the prompt callback deadlocks forever.
+func TestStdoutSecondLogDuringPauseWaitDoesNotDeadlock(t *testing.T) {
+	s := newStdout()
+
+	go s.Pause()
+	time.Sleep(50 * time.Millisecond)
+
+	go func() {
+		s.Log(time.Now(), "first log consumes pause and waits on resume")
+	}()
+	time.Sleep(50 * time.Millisecond)
+
+	secondDone := make(chan struct{})
+	go func() {
+		s.Log(time.Now(), "second log must acquire mutex while first waits for Resume")
+		close(secondDone)
+	}()
+
+	select {
+	case <-secondDone:
+	case <-time.After(2 * time.Second):
+		t.Fatal("deadlock: second Log blocked on mutex while first waits for Resume")
+	}
+
+	s.Resume()
+	time.Sleep(50 * time.Millisecond)
+}
diff --git a/internal/io/dlog/rawlog_test.go b/internal/io/dlog/rawlog_test.go
new file mode 100644
index 0000000..7fb7a05
--- /dev/null
+++ b/internal/io/dlog/rawlog_test.go
@@ -0,0 +1,76 @@
+package dlog
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/mimecast/dtail/internal/config"
+	"github.com/mimecast/dtail/internal/io/dlog/loggers"
+)
+
+// recordingLogger records whether a message arrived via the diagnostic (Log)
+// path or the payload (Raw) path, so the test can assert how RawLog vs Raw route
+// their messages. It reports SupportsColors()=false so the callers take their
+// non-color branch (logger.Log / logger.Raw), which is what the routing test
+// needs to observe.
+type recordingLogger struct {
+	mutex sync.Mutex
+	logs  []string
+	raws  []string
+}
+
+func (r *recordingLogger) Log(now time.Time, message string) {
+	r.mutex.Lock()
+	defer r.mutex.Unlock()
+	r.logs = append(r.logs, message)
+}
+func (r *recordingLogger) LogWithColors(now time.Time, message, colored string) { r.Log(now, message) }
+func (r *recordingLogger) Raw(now time.Time, message string) {
+	r.mutex.Lock()
+	defer r.mutex.Unlock()
+	r.raws = append(r.raws, message)
+}
+func (r *recordingLogger) RawWithColors(now time.Time, message, colored string) { r.Raw(now, message) }
+func (r *recordingLogger) Start(ctx context.Context, wg *sync.WaitGroup)        { wg.Done() }
+func (r *recordingLogger) Flush()                                               {}
+func (r *recordingLogger) Pause()                                               {}
+func (r *recordingLogger) Resume()                                              {}
+func (r *recordingLogger) Rotate()                                              {}
+func (r *recordingLogger) SupportsColors() bool                                 { return false }
+
+var _ loggers.Logger = (*recordingLogger)(nil)
+
+// TestRawLogUsesDiagnosticSink is the regression guard for the ReportServerError
+// footgun: a server-error audit line must go through the diagnostic (Log) sink,
+// not the payload (Raw) sink. Only the Log sink is written to the client log file
+// by default (Client.LogPayload=false gates the Raw/payload sink out of the file),
+// so a server error routed via Raw would silently vanish from the on-disk audit
+// trail. This asserts RawLog -> Log and, for contrast, Raw -> Raw.
+func TestRawLogUsesDiagnosticSink(t *testing.T) {
+	prevClient := config.Client
+	config.Client = &config.ClientConfig{TermColorsEnable: false}
+	t.Cleanup(func() { config.Client = prevClient })
+
+	rec := &recordingLogger{}
+	d := &DLog{logger: rec}
+
+	const serverError = "SERVER|srv1|ERROR|journal file targets require server capability journal-v1"
+	d.RawLog(serverError)
+
+	if len(rec.logs) != 1 || rec.logs[0] != serverError {
+		t.Fatalf("RawLog must reach the diagnostic (Log) sink verbatim; got logs=%v raws=%v",
+			rec.logs, rec.raws)
+	}
+	if len(rec.raws) != 0 {
+		t.Fatalf("RawLog must NOT use the payload (Raw) sink (it would be gated out of the file); got raws=%v",
+			rec.raws)
+	}
+
+	// Contrast: bulk payload still goes through the Raw/payload sink.
+	d.Raw("payload-line\n")
+	if len(rec.raws) != 1 || rec.raws[0] != "payload-line\n" {
+		t.Fatalf("Raw must reach the payload (Raw) sink; got raws=%v", rec.raws)
+	}
+}
diff --git a/internal/io/dlog/rotation.go b/internal/io/dlog/rotation.go
index 15ce1fd..e2a1bb8 100644
--- a/internal/io/dlog/rotation.go
+++ b/internal/io/dlog/rotation.go
@@ -12,16 +12,20 @@ import (
 func rotation(ctx context.Context) {
 	rotateCh := make(chan os.Signal, 1)
 	signal.Notify(rotateCh, syscall.SIGHUP)
-	go func() {
-		for {
-			select {
-			case <-rotateCh:
-				Common.Debug("Invoking log rotation")
-				loggers.FactoryRotate()
-				return
-			case <-ctx.Done():
-				return
-			}
+	go rotateLoop(ctx, rotateCh, loggers.FactoryRotate)
+}
+
+// rotateLoop services the log-rotation channel until ctx is cancelled. It is
+// split out from rotation so tests can drive it directly with a fake rotate
+// function and a test-owned channel.
+func rotateLoop(ctx context.Context, rotateCh <-chan os.Signal, rotate func()) {
+	for {
+		select {
+		case <-rotateCh:
+			Common.Debug("Invoking log rotation")
+			rotate()
+		case <-ctx.Done():
+			return
 		}
-	}()
+	}
 }
diff --git a/internal/io/dlog/rotation_test.go b/internal/io/dlog/rotation_test.go
new file mode 100644
index 0000000..3cc2701
--- /dev/null
+++ b/internal/io/dlog/rotation_test.go
@@ -0,0 +1,61 @@
+package dlog
+
+import (
+	"context"
+	"os"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// TestRotateLoopHandlesMultipleSignals is a regression test for a bug where
+// rotateLoop returned after the first signal, so subsequent SIGHUPs were
+// silently dropped. Sending two signals on rotateCh must result in two
+// rotate() invocations before ctx is cancelled.
+func TestRotateLoopHandlesMultipleSignals(t *testing.T) {
+	prev := Common
+	Common = &DLog{maxLevel: None}
+	t.Cleanup(func() { Common = prev })
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	rotateCh := make(chan os.Signal, 2)
+	var count int32
+	rotated := make(chan struct{}, 2)
+	rotate := func() {
+		atomic.AddInt32(&count, 1)
+		rotated <- struct{}{}
+	}
+
+	done := make(chan struct{})
+	go func() {
+		rotateLoop(ctx, rotateCh, rotate)
+		close(done)
+	}()
+
+	rotateCh <- os.Interrupt
+	waitForRotate(t, rotated, "first")
+	rotateCh <- os.Interrupt
+	waitForRotate(t, rotated, "second")
+
+	cancel()
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("rotateLoop did not return after ctx cancel")
+	}
+
+	if got := atomic.LoadInt32(&count); got != 2 {
+		t.Fatalf("rotate() called %d times, want 2", got)
+	}
+}
+
+func waitForRotate(t *testing.T, rotated <-chan struct{}, label string) {
+	t.Helper()
+	select {
+	case <-rotated:
+	case <-time.After(2 * time.Second):
+		t.Fatalf("rotate() was not invoked for %s signal", label)
+	}
+}
diff --git a/internal/io/fs/catfile.go b/internal/io/fs/catfile.go
index e4676f3..ac42fc0 100644
--- a/internal/io/fs/catfile.go
+++ b/internal/io/fs/catfile.go
@@ -6,7 +6,9 @@ type CatFile struct {
 }
 
 // NewCatFile returns a new file catter.
-func NewCatFile(filePath string, globID string, serverMessages chan<- string) CatFile {
+func NewCatFile(filePath string, globID string, serverMessages chan<- string,
+	maxLineLength int) CatFile {
+
 	return CatFile{
 		readFile: readFile{
 			filePath:       filePath,
@@ -15,6 +17,16 @@ func NewCatFile(filePath string, globID string, serverMessages chan<- string) Ca
 			retry:          false,
 			canSkipLines:   false,
 			seekEOF:        false,
+			maxLineLength:  maxLineLength,
 		},
 	}
 }
+
+// NewValidatedCatFile returns a new file catter backed by a rooted open target.
+func NewValidatedCatFile(filePath string, target ValidatedReadTarget, globID string,
+	serverMessages chan<- string, maxLineLength int) CatFile {
+
+	cat := NewCatFile(filePath, globID, serverMessages, maxLineLength)
+	cat.readFile.validatedTarget = &target
+	return cat
+}
diff --git a/internal/io/fs/filereader.go b/internal/io/fs/filereader.go
index e27d2a7..7f36a29 100644
--- a/internal/io/fs/filereader.go
+++ b/internal/io/fs/filereader.go
@@ -9,9 +9,13 @@ import (
 )
 
 // FileReader is the interface used on the dtail server to read/cat/grep/mapr...
-// a file.
+// a file. Line delivery is processor-based (line.Processor); the historic
+// channel-based Start(chan<- *line.Line) method was removed once every read path
+// migrated to the processor pipeline (task iv0).
 type FileReader interface {
-	Start(ctx context.Context, ltx lcontext.LContext, lines chan<- *line.Line,
+	StartWithProcessor(ctx context.Context, ltx lcontext.LContext, processor line.Processor,
+		re regex.Regex) error
+	StartWithProcessorOptimized(ctx context.Context, ltx lcontext.LContext, processor line.Processor,
 		re regex.Regex) error
 	FilePath() string
 	Retry() bool
diff --git a/internal/io/fs/permissions/permission.go b/internal/io/fs/permissions/permission.go
index aaab9e7..889fe9e 100644
--- a/internal/io/fs/permissions/permission.go
+++ b/internal/io/fs/permissions/permission.go
@@ -1,5 +1,4 @@
 //go:build !linuxacl
-// +build !linuxacl
 
 package permissions
 
diff --git a/internal/io/fs/permissions/permission_linuxacl.go b/internal/io/fs/permissions/permission_linuxacl.go
index bfac7e2..0a334e6 100644
--- a/internal/io/fs/permissions/permission_linuxacl.go
+++ b/internal/io/fs/permissions/permission_linuxacl.go
@@ -1,5 +1,4 @@
 //go:build linuxacl
-// +build linuxacl
 
 package permissions
 
diff --git a/internal/io/fs/permissions/permission_test.go b/internal/io/fs/permissions/permission_test.go
index e28c67b..3d9f8e3 100644
--- a/internal/io/fs/permissions/permission_test.go
+++ b/internal/io/fs/permissions/permission_test.go
@@ -1,5 +1,4 @@
 //go:build linuxacl
-// +build linuxacl
 
 package permissions
 
diff --git a/internal/io/fs/readfile.go b/internal/io/fs/readfile.go
index dc1d8ea..7969cf0 100644
--- a/internal/io/fs/readfile.go
+++ b/internal/io/fs/readfile.go
@@ -2,7 +2,6 @@ package fs
 
 import (
 	"bufio"
-	"bytes"
 	"compress/gzip"
 	"context"
 	"errors"
@@ -10,25 +9,18 @@ import (
 	"io"
 	"os"
 	"strings"
-	"sync"
 	"time"
 
-	"github.com/mimecast/dtail/internal/config"
 	"github.com/mimecast/dtail/internal/io/dlog"
-	"github.com/mimecast/dtail/internal/io/line"
-	"github.com/mimecast/dtail/internal/io/pool"
-	"github.com/mimecast/dtail/internal/lcontext"
-	"github.com/mimecast/dtail/internal/regex"
-
-	"github.com/DataDog/zstd"
 )
 
 type readStatus int
 
 const (
-	nothing         readStatus = iota
-	abortReading    readStatus = iota
-	continueReading readStatus = iota
+	nothing              readStatus = iota
+	abortReading         readStatus = iota
+	continueReading      readStatus = iota
+	defaultMaxLineLength            = 1024 * 1024
 )
 
 // Used to tail and filter a local log file.
@@ -37,6 +29,8 @@ type readFile struct {
 	stats
 	// Path of log file to tail.
 	filePath string
+	// Rooted target used for validated server-side re-opens.
+	validatedTarget *ValidatedReadTarget
 	// The glob identifier of the file.
 	globID string
 	// Channel to send a server message to the dtail client
@@ -49,6 +43,8 @@ type readFile struct {
 	seekEOF bool
 	// Warned already about a long line.
 	warnedAboutLongLine bool
+	// Maximum line length before a line is split.
+	maxLineLength int
 }
 
 // String returns the string representation of the readFile
@@ -72,52 +68,42 @@ func (f readFile) Retry() bool {
 	return f.retry
 }
 
-// Start tailing a log file.
-func (f readFile) Start(ctx context.Context, ltx lcontext.LContext,
-	lines chan<- *line.Line, re regex.Regex) error {
-
-	reader, fd, err := f.makeReader()
-	if fd != nil {
-		defer fd.Close()
+func (f *readFile) lineLimit() int {
+	if f.maxLineLength <= 0 {
+		return defaultMaxLineLength
 	}
-	if err != nil {
-		return err
-	}
-
-	rawLines := make(chan *bytes.Buffer, 100)
-	truncate := make(chan struct{})
-
-	readCtx, readCancel := context.WithCancel(ctx)
-	var filterWg sync.WaitGroup
-	filterWg.Add(1)
+	return f.maxLineLength
+}
 
-	go f.periodicTruncateCheck(ctx, truncate)
-	go func() {
-		f.filter(ctx, ltx, rawLines, lines, re)
-		filterWg.Done()
-		// If the filter stopped, make the reader stop too, no need to read
-		// more data if there is nothing more the filter wants to filter for!
-		// E.g. it could be that we only want to filter N matches but not more.
-		readCancel()
-	}()
+func (f *readFile) warnAboutLongLine(ctx context.Context) bool {
+	if f.warnedAboutLongLine {
+		return true
+	}
 
-	err = f.read(readCtx, fd, reader, rawLines, truncate)
-	close(rawLines)
-	// Filter may sends some data still. So wait until it is done here.
-	filterWg.Wait()
+	if f.serverMessages == nil {
+		f.warnedAboutLongLine = true
+		return true
+	}
 
-	return err
+	select {
+	case f.serverMessages <- dlog.Common.Warn(f.filePath,
+		"Long log line, splitting into multiple lines") + "\n":
+		f.warnedAboutLongLine = true
+		return true
+	case <-ctx.Done():
+		return false
+	}
 }
 
-func (f *readFile) makeReader() (*bufio.Reader, *os.File, error) {
+func (f *readFile) makeReader() (*bufio.Reader, *os.File, io.Closer, error) {
 	if f.filePath == "" && f.globID == "-" {
 		return f.makePipeReader()
 	}
 	return f.makeFileReader()
 }
 
-func (f *readFile) makeFileReader() (reader *bufio.Reader, fd *os.File, err error) {
-	if fd, err = os.Open(f.filePath); err != nil {
+func (f *readFile) makeFileReader() (reader *bufio.Reader, fd *os.File, decompressor io.Closer, err error) {
+	if fd, err = f.openFile(); err != nil {
 		return
 	}
 
@@ -127,21 +113,32 @@ func (f *readFile) makeFileReader() (reader *bufio.Reader, fd *os.File, err erro
 		}
 	}
 
-	reader, err = f.makeCompressedFileReader(fd)
+	reader, decompressor, err = f.makeCompressedFileReader(fd)
 	return
 }
 
-func (f *readFile) makePipeReader() (*bufio.Reader, *os.File, error) {
-	return bufio.NewReader(os.Stdin), nil, nil
+func (f *readFile) openFile() (*os.File, error) {
+	if f.validatedTarget != nil {
+		return f.validatedTarget.Open()
+	}
+	return os.Open(f.filePath)
 }
 
-func (f *readFile) periodicTruncateCheck(ctx context.Context, truncate chan struct{}) {
+func (f *readFile) makePipeReader() (*bufio.Reader, *os.File, io.Closer, error) {
+	return bufio.NewReader(os.Stdin), nil, nil, nil
+}
+
+func (f *readFile) periodicTruncateCheck(ctx context.Context, truncate chan<- struct{}) {
+	ticker := time.NewTicker(time.Second * 3)
+	defer ticker.Stop()
+
 	for {
 		select {
-		case <-time.After(time.Second * 3):
+		case <-ticker.C:
 			select {
 			case truncate <- struct{}{}:
 			case <-ctx.Done():
+				return
 			}
 		case <-ctx.Done():
 			return
@@ -149,7 +146,7 @@ func (f *readFile) periodicTruncateCheck(ctx context.Context, truncate chan st