feat: DTail fork — server/client feature development

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>

1 files changed, 430 insertions, 0 deletions

diff --git a/internal/io/fs/readfile_processor_optimized.go b/internal/io/fs/readfile_processor_optimized.go
new file mode 100644
index 0000000..6426962
--- /dev/null
+++ b/internal/io/fs/readfile_processor_optimized.go
@@ -0,0 +1,430 @@
+package fs
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"io"
+	"os"
+	"time"
+
+	"github.com/mimecast/dtail/internal/ctxutil"
+	"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"
+)
+
+// readWithProcessorOptimized reads from the file using buffered line reading
+// instead of byte-by-byte reading for better performance
+func (f *readFile) readWithProcessorOptimized(ctx context.Context, fd *os.File, reader *bufio.Reader,
+	truncate <-chan struct{}, ltx lcontext.LContext, processor line.Processor, re regex.Regex) error {
+
+	// Create a line filter processor that wraps the given processor
+	filterProcessor := &filteringProcessor{
+		processor: processor,
+		re:        re,
+		ltx:       ltx,
+		stats:     &f.stats,
+		globID:    f.globID,
+	}
+
+	// Compute the local-context predicate once. When no context is requested we
+	// can take the zero-copy fast path (match before copy); when it is, every
+	// line must be buffered so surrounding before/after lines remain available.
+	hasContext := ltx.Has()
+
+	// Use a scanner for efficient line reading
+	scanner := bufio.NewScanner(reader)
+
+	// Get a buffer from the pool instead of allocating a new one
+	bufPtr := pool.GetScannerBuffer()
+	buf := *bufPtr
+	maxTokenSize := 1024 * 1024 // 1MB max token size
+	scanner.Buffer(buf, maxTokenSize)
+
+	// Ensure we return the buffer to the pool when done
+	defer pool.PutScannerBuffer(bufPtr)
+
+	// Use the cancellation-aware split function so long-line warnings can be
+	// abandoned if the caller cancels while the reader is blocked.
+	scanner.Split(func(data []byte, atEOF bool) (advance int, token []byte, err error) {
+		return f.scanLinesWithMaxLength(ctx, data, atEOF)
+	})
+
+	for scanner.Scan() {
+		// Check context cancellation
+		select {
+		case <-ctx.Done():
+			return nil
+		default:
+		}
+
+		// Check for file truncation. The periodicTruncateCheck goroutine
+		// (started in StartWithProcessorOptimized) already ticks every 3s and
+		// signals on the unbuffered truncate channel, so this non-blocking
+		// receive only re-stats the file on that cadence. Keeping the timing in
+		// the goroutine lets the per-line cost be a single atomic load on an
+		// empty channel (Go's non-blocking chanrecv fast path) instead of a
+		// per-line time.Since/runtime.nanotime call, which profiling showed as
+		// 10-18% of serverless dcat/dgrep CPU. This path is non-follow
+		// (cat/grep) only; follow mode uses tailWithProcessorOptimized, which
+		// has its own truncate handling.
+		select {
+		case <-truncate:
+			if isTruncated, err := f.truncated(fd); isTruncated {
+				return err
+			}
+		default:
+		}
+
+		// Get the line data. scanner.Bytes() is only valid until the next
+		// Scan(); we must not retain it across iterations.
+		lineData := scanner.Bytes()
+		f.updatePosition()
+
+		if !hasContext {
+			// Fast path: run the regex on the scanner's slice directly and only
+			// copy into a pooled buffer on a match. At low hit rates this skips
+			// the pool.Get + copy for the discarded (non-matching) lines.
+			if err := filterProcessor.ProcessFilteredRaw(lineData); err != nil {
+				if isEarlyStop(err) {
+					return nil
+				}
+				return err
+			}
+			continue
+		}
+
+		// Local-context path: buffer every line (before/after context needs the
+		// surrounding non-matching lines), so copy into a pooled buffer first.
+		lineBuf := pool.BytesBuffer.Get().(*bytes.Buffer)
+		lineBuf.Write(lineData)
+		if err := filterProcessor.ProcessFilteredLine(lineBuf); err != nil {
+			if isEarlyStop(err) {
+				return nil
+			}
+			return err
+		}
+	}
+
+	// Check for scanner errors
+	if err := scanner.Err(); err != nil {
+		// Handle EOF specially for tailing
+		if err == io.EOF && f.seekEOF {
+			// For tail mode, we want to keep reading
+			return nil
+		}
+		return err
+	}
+
+	return nil
+}
+
+// isEarlyStop reports whether err is the io.EOF sentinel that filteringProcessor
+// returns from processWithContext once a max-count (-m/-max) limit is reached.
+// This is a NORMAL early stop, not a genuine I/O error: bufio.Scanner signals
+// real end-of-input via Scan()==false and never returns io.EOF from
+// ProcessFiltered*, so any io.EOF bubbling up from the filter can only be the
+// max-count sentinel. The byte-by-byte path (readWithProcessor) swallows it and
+// returns nil; the optimized path must do the same, otherwise the sentinel leaks
+// out to the caller and is logged as a spurious SERVER|...|ERROR|...|EOF line.
+// Real (non-EOF) processor errors are left untouched so they still surface.
+//
+// Bare equality (== io.EOF) is intentional and must NOT become errors.Is: the
+// sentinel is returned bare by processWithContext, so exact identity matches it
+// precisely. A WRAPPED io.EOF, by contrast, can only originate from a genuine
+// downstream failure (e.g. an ssh channel Write after the peer closed), which we
+// deliberately do NOT want to mistake for a clean early stop.
+func isEarlyStop(err error) bool {
+	return err == io.EOF
+}
+
+// scanLinesPreserveEndings is a custom split function that preserves original line endings
+// and respects MaxLineLength
+func (f *readFile) scanLinesPreserveEndings(data []byte, atEOF bool) (advance int, token []byte, err error) {
+	if atEOF && len(data) == 0 {
+		return 0, nil, nil
+	}
+
+	maxLineLen := f.lineLimit()
+
+	// Look for a newline
+	if i := bytes.IndexByte(data, '\n'); i >= 0 {
+		// Check if the line before the newline exceeds max length
+		if i > maxLineLen {
+			// Line is too long, split it silently at maxLineLen
+			return maxLineLen, data[0:maxLineLen], nil
+		}
+
+		// Line is within limit, include the line ending in the token
+		// Check if there's a \r before the \n
+		if i > 0 && data[i-1] == '\r' {
+			// Windows line ending (\r\n) - include both in token
+			return i + 1, data[0 : i+1], nil
+		}
+		// Unix line ending (\n) - include it in token
+		return i + 1, data[0 : i+1], nil
+	}
+
+	// If we're at EOF, we have a final, non-terminated line
+	if atEOF {
+		if len(data) > maxLineLen {
+			// Even at EOF, respect max line length (split silently)
+			return maxLineLen, data[0:maxLineLen], nil
+		}
+		return len(data), data, nil
+	}
+
+	// If the line is too long, split it
+	if len(data) >= maxLineLen {
+		// Return a chunk up to MaxLineLength (split silently)
+		return maxLineLen, data[0:maxLineLen], nil
+	}
+
+	// Request more data
+	return 0, nil, nil
+}
+
+// scanLinesWithMaxLength is a custom split function for bufio.Scanner that respects MaxLineLength.
+// It is kept context-aware so long-line warnings can still be dropped when the reader is canceled.
+func (f *readFile) scanLinesWithMaxLength(ctx context.Context, data []byte, atEOF bool) (advance int, token []byte, err error) {
+	if atEOF && len(data) == 0 {
+		return 0, nil, nil
+	}
+
+	maxLineLen := f.lineLimit()
+
+	// Look for a newline
+	if i := bytes.IndexByte(data, '\n'); i >= 0 {
+		// Check if the line before the newline exceeds max length
+		if i > maxLineLen {
+			// Line is too long, split it at maxLineLen
+			if !f.warnAboutLongLine(ctx) {
+				return 0, nil, ctx.Err()
+			}
+			return maxLineLen, data[0:maxLineLen], nil
+		}
+		// We have a full line within the limit
+		f.warnedAboutLongLine = false // Reset warning for next long line sequence
+		return i + 1, data[0 : i+1], nil
+	}
+
+	// If we're at EOF, we have a final, non-terminated line
+	if atEOF {
+		if len(data) > maxLineLen {
+			// Even at EOF, respect max line length
+			if !f.warnAboutLongLine(ctx) {
+				return 0, nil, ctx.Err()
+			}
+			return maxLineLen, data[0:maxLineLen], nil
+		}
+		return len(data), data, nil
+	}
+
+	// If the line is too long, split it
+	if len(data) >= maxLineLen {
+		// Warn about long line (only once)
+		if !f.warnAboutLongLine(ctx) {
+			return 0, nil, ctx.Err()
+		}
+
+		// Return a chunk up to MaxLineLength
+		return maxLineLen, data[0:maxLineLen], nil
+	}
+
+	// Request more data
+	return 0, nil, nil
+}
+
+// StartWithProcessorOptimized starts reading a log file using an optimized LineProcessor implementation.
+// This version uses buffered line reading instead of byte-by-byte reading.
+func (f *readFile) StartWithProcessorOptimized(ctx context.Context, ltx lcontext.LContext,
+	processor line.Processor, re regex.Regex) error {
+
+	reader, fd, decompressor, err := f.makeReader()
+	if fd != nil {
+		defer fd.Close()
+	}
+	if decompressor != nil {
+		defer func() {
+			if closeErr := decompressor.Close(); closeErr != nil {
+				dlog.Common.Warn(f.filePath, "Unable to close compressed reader", closeErr)
+			}
+		}()
+	}
+	if err != nil {
+		return err
+	}
+
+	// Create a cancelable context for the truncate check goroutine
+	truncateCtx, cancelTruncate := context.WithCancel(ctx)
+	defer cancelTruncate()
+
+	truncate := make(chan struct{})
+
+	go f.periodicTruncateCheck(truncateCtx, truncate)
+
+	// For tail mode, we need to handle continuous reading
+	if f.seekEOF {
+		return f.tailWithProcessorOptimized(ctx, fd, reader, truncate, ltx, processor, re)
+	}
+
+	// For cat/grep mode, just read once
+	err = f.readWithProcessorOptimized(ctx, fd, reader, truncate, ltx, processor, re)
+
+	// Ensure any buffered data is flushed
+	if flushErr := processor.Flush(); flushErr != nil && err == nil {
+		err = flushErr
+	}
+
+	return err
+}
+
+// tailWithProcessorOptimized handles continuous reading for tail mode
+func (f *readFile) tailWithProcessorOptimized(ctx context.Context, fd *os.File, reader *bufio.Reader,
+	truncate <-chan struct{}, ltx lcontext.LContext, processor line.Processor, re regex.Regex) error {
+
+	// Create a line filter processor
+	filterProcessor := &filteringProcessor{
+		processor: processor,
+		re:        re,
+		ltx:       ltx,
+		stats:     &f.stats,
+		globID:    f.globID,
+	}
+
+	// Compute the local-context predicate once (see readWithProcessorOptimized):
+	// without context we take the zero-copy match-before-copy fast path.
+	hasContext := ltx.Has()
+
+	// Buffer for partial lines
+	partialLine := pool.BytesBuffer.Get().(*bytes.Buffer)
+	defer pool.RecycleBytesBuffer(partialLine)
+
+	// Get a buffer from the pool for reading
+	bufPtr := pool.GetMediumBuffer()
+	defer pool.PutMediumBuffer(bufPtr)
+
+	// processPartialLine advances the line position and hands the currently
+	// accumulated partialLine to the filter. Without local context it takes the
+	// zero-copy fast path (match on partialLine.Bytes(), copy only on a match);
+	// with context it copies into a pooled buffer so surrounding lines stay
+	// buffered. partialLine is owned by this loop (reset after each call), so the
+	// fast path never retains its slice past the copy-on-match.
+	processPartialLine := func() error {
+		f.updatePosition()
+		if !hasContext {
+			return filterProcessor.ProcessFilteredRaw(partialLine.Bytes())
+		}
+		lineBuf := pool.BytesBuffer.Get().(*bytes.Buffer)
+		lineBuf.Write(partialLine.Bytes())
+		return filterProcessor.ProcessFilteredLine(lineBuf)
+	}
+
+	for {
+		// Read available data using pooled buffer
+		buf := (*bufPtr)[:cap(*bufPtr)] // Reset to full capacity
+		n, err := reader.Read(buf)
+
+		if n > 0 {
+			// Process the data we read
+			data := buf[:n]
+
+			// Process complete lines
+			for len(data) > 0 {
+				// Find newline
+				idx := bytes.IndexByte(data, '\n')
+
+				if idx >= 0 {
+					// Complete line found
+					partialLine.Write(data[:idx])
+
+					// Process the line if it's not empty
+					if partialLine.Len() > 0 {
+						if err := processPartialLine(); err != nil {
+							// Max-count early stop is a clean stop, not an error
+							// (see isEarlyStop); mirror the byte-by-byte path.
+							if isEarlyStop(err) {
+								return nil
+							}
+							return err
+						}
+					}
+
+					partialLine.Reset()
+					data = data[idx+1:]
+
+					// Reset long line warning
+					f.warnedAboutLongLine = false
+				} else {
+					// No newline, add to partial line
+					partialLine.Write(data)
+
+					// Check if line is too long
+					if partialLine.Len() >= f.lineLimit() {
+						if !f.warnAboutLongLine(ctx) {
+							return nil
+						}
+
+						// Process the partial line
+						if err := processPartialLine(); err != nil {
+							if isEarlyStop(err) {
+								return nil
+							}
+							return err
+						}
+
+						partialLine.Reset()
+					}
+
+					break
+				}
+			}
+
+			// Flush processor periodically
+			if err := processor.Flush(); err != nil {
+				return err
+			}
+		}
+
+		// Handle read errors
+		if err != nil {
+			if err != io.EOF {
+				return err
+			}
+
+			waitForMoreData := true
+
+			// EOF handling
+			select {
+			case <-ctx.Done():
+				return nil
+			case <-truncate:
+				if isTruncated, err := f.truncated(fd); isTruncated {
+					return err
+				}
+				waitForMoreData = false
+			default:
+			}
+
+			if waitForMoreData && !ctxutil.Sleep(ctx, 100*time.Millisecond) {
+				return nil
+			}
+		}
+
+		// Check for cancellation
+		select {
+		case <-ctx.Done():
+			// Process any remaining partial line
+			if partialLine.Len() > 0 {
+				if err := processPartialLine(); err != nil && !isEarlyStop(err) {
+					return err
+				}
+			}
+			return nil
+		default:
+		}
+	}
+}