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>

17 files changed, 2297 insertions, 402 deletions

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 stru
 	}
 }
 
-func (f *readFile) makeCompressedFileReader(fd *os.File) (reader *bufio.Reader, err error) {
+func (f *readFile) makeCompressedFileReader(fd *os.File) (reader *bufio.Reader, decompressor io.Closer, err error) {
 	switch {
 	case strings.HasSuffix(f.FilePath(), ".gz"):
 		fallthrough
@@ -160,82 +157,16 @@ func (f *readFile) makeCompressedFileReader(fd *os.File) (reader *bufio.Reader,
 		if err != nil {
 			return
 		}
+		decompressor = gzipReader
 		reader = bufio.NewReader(gzipReader)
 	case strings.HasSuffix(f.FilePath(), ".zst"):
-		dlog.Common.Info(f.FilePath(), "Detected zstd compression format")
-		reader = bufio.NewReader(zstd.NewReader(fd))
+		return f.makeZstdReader(fd)
 	default:
 		reader = bufio.NewReader(fd)
 	}
 	return
 }
 
-func (f *readFile) read(ctx context.Context, fd *os.File, reader *bufio.Reader,
-	rawLines chan *bytes.Buffer, truncate <-chan struct{}) error {
-
-	var offset uint64
-	message := pool.BytesBuffer.Get().(*bytes.Buffer)
-
-	for {
-		b, err := reader.ReadByte()
-		if err != nil {
-			status, err := f.handleReadError(ctx, err, fd, rawLines, truncate, message)
-			if abortReading == status {
-				return err
-			}
-			time.Sleep(time.Millisecond * 100)
-			continue
-		}
-
-		offset++
-		message.WriteByte(b)
-
-		status, newMessage := f.handleReadByte(ctx, b, rawLines, message)
-		if status == abortReading {
-			return nil
-		}
-		message = newMessage
-	}
-}
-
-// Filter log lines matching a given regular expression.
-func (f *readFile) filter(ctx context.Context, ltx lcontext.LContext,
-	rawLines <-chan *bytes.Buffer, lines chan<- *line.Line, re regex.Regex) {
-
-	// Do we have any kind of local context settings? If so then run the more complex
-	// filterWithLContext method.
-	if ltx.Has() {
-		// We can not skip transmitting any lines to the client with a local
-		// grep context specified.
-		f.canSkipLines = false
-		f.filterWithLContext(ctx, ltx, rawLines, lines, re)
-		return
-	}
-
-	f.filterWithoutLContext(ctx, rawLines, lines, re)
-}
-
-func (f *readFile) transmittable(rawLine *bytes.Buffer, length, capacity int,
-	re regex.Regex) (*line.Line, bool) {
-
-	newLine := line.Null()
-	if !re.Match(rawLine.Bytes()) {
-		f.updateLineNotMatched()
-		f.updateLineNotTransmitted()
-		return newLine, false
-	}
-	f.updateLineMatched()
-
-	// Can we actually send more messages, channel capacity reached?
-	if f.canSkipLines && length >= capacity {
-		f.updateLineNotTransmitted()
-		return newLine, false
-	}
-	f.updateLineTransmitted()
-
-	return line.New(rawLine, f.totalLineCount(), f.transmittedPerc(), f.globID), true
-}
-
 // Check wether log file is truncated. Returns nil if not.
 func (f *readFile) truncated(fd *os.File) (bool, error) {
 	if fd == nil {
@@ -250,7 +181,7 @@ func (f *readFile) truncated(fd *os.File) (bool, error) {
 		return true, err
 	}
 	// Can not open file at original path.
-	pathFd, err := os.Open(f.filePath)
+	pathFd, err := f.openFile()
 	if err != nil {
 		return true, err
 	}
@@ -267,68 +198,3 @@ func (f *readFile) truncated(fd *os.File) (bool, error) {
 	return false, nil
 }
 
-// Deal with the scenario that nothing could be read from the fd.
-func (f *readFile) handleReadError(ctx context.Context, err error, fd *os.File,
-	rawLines chan *bytes.Buffer, truncate <-chan struct{},
-	message *bytes.Buffer) (readStatus, error) {
-
-	if err != io.EOF {
-		return abortReading, err
-	}
-
-	select {
-	case <-truncate:
-		if isTruncated, err := f.truncated(fd); isTruncated {
-			return abortReading, err
-		}
-	case <-ctx.Done():
-		return abortReading, nil
-	default:
-	}
-
-	if !f.seekEOF {
-		dlog.Common.Info(f.FilePath(), "End of file reached")
-		if len(message.Bytes()) > 0 {
-			select {
-			case rawLines <- message:
-			case <-ctx.Done():
-			}
-		}
-		return abortReading, nil
-	}
-
-	return nothing, nil
-}
-
-// Now process the byte we just read from the fd.
-func (f *readFile) handleReadByte(ctx context.Context, b byte,
-	rawLines chan *bytes.Buffer, message *bytes.Buffer) (readStatus, *bytes.Buffer) {
-
-	switch b {
-	case '\n':
-		select {
-		case rawLines <- message:
-			message = pool.BytesBuffer.Get().(*bytes.Buffer)
-			f.warnedAboutLongLine = false
-		case <-ctx.Done():
-			return abortReading, message
-		}
-	default:
-		if message.Len() >= config.Server.MaxLineLength {
-			if !f.warnedAboutLongLine {
-				f.serverMessages <- dlog.Common.Warn(f.filePath,
-					"Long log line, splitting into multiple lines") + "\n"
-				f.warnedAboutLongLine = true
-			}
-			message.WriteByte('\n')
-			select {
-			case rawLines <- message:
-				message = pool.BytesBuffer.Get().(*bytes.Buffer)
-			case <-ctx.Done():
-				return abortReading, message
-			}
-		}
-	}
-
-	return nothing, message
-}
diff --git a/internal/io/fs/readfile_nozstd.go b/internal/io/fs/readfile_nozstd.go
new file mode 100644
index 0000000..afd4523
--- /dev/null
+++ b/internal/io/fs/readfile_nozstd.go
@@ -0,0 +1,16 @@
+//go:build nozstd
+
+package fs
+
+import (
+	"bufio"
+	"fmt"
+	"io"
+	"os"
+)
+
+func (f *readFile) makeZstdReader(fd *os.File) (reader *bufio.Reader, decompressor io.Closer, err error) {
+	_ = fd
+	err = fmt.Errorf("%s: zstd is not supported in this build (built with -tags nozstd)", f.FilePath())
+	return
+}
diff --git a/internal/io/fs/readfile_processor.go b/internal/io/fs/readfile_processor.go
new file mode 100644
index 0000000..37809e0
--- /dev/null
+++ b/internal/io/fs/readfile_processor.go
@@ -0,0 +1,359 @@
+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"
+)
+
+// StartWithProcessor starts reading a log file using a LineProcessor for handling lines.
+// This is a channel-less implementation for better performance.
+func (f *readFile) StartWithProcessor(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
+	}
+
+	truncateCtx, cancelTruncate := context.WithCancel(ctx)
+	defer cancelTruncate()
+
+	truncate := make(chan struct{})
+
+	go f.periodicTruncateCheck(truncateCtx, truncate)
+
+	// Process file with direct callbacks instead of channels
+	err = f.readWithProcessor(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
+}
+
+// readWithProcessor reads from the file and processes lines directly without channels
+func (f *readFile) readWithProcessor(ctx context.Context, fd *os.File, reader *bufio.Reader,
+	truncate <-chan struct{}, ltx lcontext.LContext, processor line.Processor, re regex.Regex) error {
+
+	var offset uint64
+	message := pool.BytesBuffer.Get().(*bytes.Buffer)
+	// Use a closure so that the CURRENT value of `message` is recycled on
+	// return, not the pointer captured at defer-registration time. Downstream
+	// code paths that take ownership of the buffer set `message = nil` before
+	// reassigning or returning, preventing a double-recycle.
+	defer func() {
+		if message != nil {
+			pool.RecycleBytesBuffer(message)
+		}
+	}()
+
+	// Create a line filter processor that wraps the given processor
+	filterProcessor := &filteringProcessor{
+		processor: processor,
+		re:        re,
+		ltx:       ltx,
+		stats:     &f.stats,
+		globID:    f.globID,
+	}
+
+	for {
+		b, err := reader.ReadByte()
+		if err != nil {
+			// handleReadErrorProcessor may hand `message` to ProcessFilteredLine
+			// (which takes ownership); in that case it sets *messagePtr = nil so
+			// the caller's defer does not recycle an already-recycled buffer.
+			status, err := f.handleReadErrorProcessor(ctx, err, fd, truncate, &message, filterProcessor)
+			if abortReading == status {
+				return err
+			}
+			if !ctxutil.Sleep(ctx, 100*time.Millisecond) {
+				return nil
+			}
+			continue
+		}
+
+		offset++
+		message.WriteByte(b)
+
+		status := f.handleReadByteProcessor(ctx, b, message, filterProcessor)
+		if status == abortReading {
+			// ProcessFilteredLine took ownership; avoid defer double-recycle.
+			message = nil
+			return nil
+		}
+		if status == continueReading {
+			// Previous buffer was consumed by ProcessFilteredLine; acquire a fresh one.
+			message = pool.BytesBuffer.Get().(*bytes.Buffer)
+		}
+	}
+}
+
+// handleReadByteProcessor processes a byte read from the file
+func (f *readFile) handleReadByteProcessor(ctx context.Context, b byte,
+	message *bytes.Buffer, processor *filteringProcessor) readStatus {
+
+	switch b {
+	case '\n':
+		// Process the complete line
+		f.updatePosition()
+		if err := processor.ProcessFilteredLine(message); err != nil {
+			return abortReading
+		}
+
+		f.warnedAboutLongLine = false
+		return continueReading
+
+	default:
+		if message.Len() >= f.lineLimit() {
+			if !f.warnAboutLongLine(ctx) {
+				return abortReading
+			}
+			// Force a line break
+			message.WriteByte('\n')
+
+			// Process the line
+			f.updatePosition()
+			if err := processor.ProcessFilteredLine(message); err != nil {
+				return abortReading
+			}
+			return continueReading
+		}
+	}
+
+	return nothing
+}
+
+// handleReadErrorProcessor handles read errors in processor mode. When it hands
+// the buffer to ProcessFilteredLine it nils out *messagePtr, signalling to the
+// caller that ownership has been transferred downstream.
+func (f *readFile) handleReadErrorProcessor(ctx context.Context, err error, fd *os.File,
+	truncate <-chan struct{}, messagePtr **bytes.Buffer, processor *filteringProcessor) (readStatus, error) {
+
+	if err != io.EOF {
+		return abortReading, err
+	}
+
+	select {
+	case <-truncate:
+		if isTruncated, err := f.truncated(fd); isTruncated {
+			return abortReading, err
+		}
+	case <-ctx.Done():
+		return abortReading, nil
+	default:
+	}
+
+	if !f.seekEOF {
+		dlog.Common.Info(f.FilePath(), "End of file reached")
+		message := *messagePtr
+		if len(message.Bytes()) > 0 {
+			// Process the last line if it doesn't end with newline.
+			f.updatePosition()
+			*messagePtr = nil
+			if processErr := processor.ProcessFilteredLine(message); processErr != nil {
+				return abortReading, processErr
+			}
+		}
+		return abortReading, nil
+	}
+
+	return nothing, nil
+}
+
+// filteringProcessor wraps a LineProcessor to add regex filtering
+type filteringProcessor struct {
+	processor line.Processor
+	re        regex.Regex
+	ltx       lcontext.LContext
+	stats     *stats
+	globID    string
+
+	// For local context handling
+	beforeBuf  []*bytes.Buffer
+	afterCount int
+	maxCount   int
+	maxReached bool
+}
+
+// ProcessFilteredLine applies regex filtering before passing to the underlying processor
+func (fp *filteringProcessor) ProcessFilteredLine(rawLine *bytes.Buffer) error {
+	// Update stats
+	lineNum := fp.stats.totalLineCount()
+
+	// Simple case: no local context
+	if !fp.ltx.Has() {
+		if !fp.re.Match(rawLine.Bytes()) {
+			fp.stats.updateLineNotMatched()
+			fp.stats.updateLineNotTransmitted()
+			pool.RecycleBytesBuffer(rawLine)
+			return nil
+		}
+
+		fp.stats.updateLineMatched()
+		fp.stats.updateLineTransmitted()
+
+		// Process the line. Per the line.Processor contract (processor.go),
+		// ownership of rawLine transfers to the processor, which recycles it on
+		// every return path. The only processors on the fs read path -
+		// DirectLineProcessor and AggregateProcessor - recycle unconditionally,
+		// even when ProcessLine returns a write error (e.g. a client disconnect /
+		// broken pipe). Recycling here on error would Put the same buffer into the
+		// shared pool.BytesBuffer a second time; the pool would then hand one object
+		// to two Get callers whose concurrent writes race and corrupt data. So do
+		// not recycle rawLine here.
+		return fp.processor.ProcessLine(rawLine, lineNum, fp.globID)
+	}
+
+	// Complex case: handle local context (before/after/max)
+	return fp.processWithContext(rawLine, lineNum)
+}
+
+// ProcessFilteredRaw is the zero-copy fast path for the no-local-context case.
+// It runs the regex match directly on the scanner-owned byte slice and only
+// acquires+fills a pooled buffer when the line actually matches. At low hit
+// rates this avoids a pool.Get + copy + pool.Put for the (vast majority of)
+// non-matching lines, which profiling showed as ~10-15% of serverless
+// dgrep CPU (sync.Pool Get/Put + bytes.Buffer.Write).
+//
+// Semantics are identical to the !ltx.Has() branch of ProcessFilteredLine: the
+// same regex, the same stats bookkeeping, and the same lineNum are used, so
+// output is byte-identical. It MUST only be called when fp.ltx.Has() is false;
+// the local-context path deliberately buffers non-matching lines (before/after
+// context) and cannot skip the copy.
+//
+// The caller passes raw = scanner.Bytes(), which is only valid until the next
+// Scan(). On a match we copy it into a pooled buffer before returning, so the
+// buffer handed to the underlying processor is a stable copy and never aliases
+// the scanner's transient slice.
+func (fp *filteringProcessor) ProcessFilteredRaw(raw []byte) error {
+	lineNum := fp.stats.totalLineCount()
+
+	if !fp.re.Match(raw) {
+		fp.stats.updateLineNotMatched()
+		fp.stats.updateLineNotTransmitted()
+		// No buffer was acquired, so there is nothing to recycle.
+		return nil
+	}
+
+	fp.stats.updateLineMatched()
+	fp.stats.updateLineTransmitted()
+
+	// Only now, on a confirmed match, pay for the buffer and the copy.
+	lineBuf := pool.BytesBuffer.Get().(*bytes.Buffer)
+	lineBuf.Write(raw)
+
+	// Ownership of lineBuf transfers to the processor, which recycles it on every
+	// return path (see ProcessFilteredLine for the full rationale). Recycling here
+	// on error would return the same buffer to the shared pool a second time and
+	// race, so leave it to the processor.
+	return fp.processor.ProcessLine(lineBuf, lineNum, fp.globID)
+}
+
+// processWithContext handles lines when local context is enabled
+func (fp *filteringProcessor) processWithContext(rawLine *bytes.Buffer, lineNum uint64) error {
+	matched := fp.re.Match(rawLine.Bytes())
+
+	if !matched {
+		fp.stats.updateLineNotMatched()
+
+		// Handle after context
+		if fp.ltx.AfterContext > 0 && fp.afterCount > 0 {
+			fp.afterCount--
+			fp.stats.updateLineTransmitted()
+			// Ownership transfers to the processor, which recycles rawLine on every
+			// return path; recycling here on error would double Put into the shared
+			// pool and race (see ProcessFilteredLine).
+			return fp.processor.ProcessLine(rawLine, lineNum, fp.globID)
+		}
+
+		// Handle before context buffer
+		if fp.ltx.BeforeContext > 0 {
+			// Add to before buffer
+			if len(fp.beforeBuf) >= fp.ltx.BeforeContext {
+				// Recycle oldest buffer
+				pool.RecycleBytesBuffer(fp.beforeBuf[0])
+				fp.beforeBuf = fp.beforeBuf[1:]
+			}
+			fp.beforeBuf = append(fp.beforeBuf, rawLine)
+		} else {
+			pool.RecycleBytesBuffer(rawLine)
+		}
+
+		fp.stats.updateLineNotTransmitted()
+		return nil
+	}
+
+	// Line matched
+	fp.stats.updateLineMatched()
+
+	// Check if we've reached max count
+	if fp.maxReached {
+		pool.RecycleBytesBuffer(rawLine)
+		return io.EOF // Stop processing
+	}
+
+	// Process before context
+	if fp.ltx.BeforeContext > 0 && len(fp.beforeBuf) > 0 {
+		for i, buf := range fp.beforeBuf {
+			fp.stats.updateLineTransmitted()
+			if err := fp.processor.ProcessLine(buf, lineNum-uint64(len(fp.beforeBuf)-i), fp.globID); err != nil {
+				// Clean up remaining buffers
+				for j := i + 1; j < len(fp.beforeBuf); j++ {
+					pool.RecycleBytesBuffer(fp.beforeBuf[j])
+				}
+				pool.RecycleBytesBuffer(rawLine)
+				return err
+			}
+		}
+		fp.beforeBuf = fp.beforeBuf[:0] // Clear the buffer
+	}
+
+	// Process the matched line. Ownership transfers to the processor, which
+	// recycles rawLine on every return path; recycling here on error would double
+	// Put into the shared pool and race (see ProcessFilteredLine).
+	fp.stats.updateLineTransmitted()
+	if err := fp.processor.ProcessLine(rawLine, lineNum, fp.globID); err != nil {
+		return err
+	}
+
+	// Update max count
+	if fp.ltx.MaxCount > 0 {
+		fp.maxCount++
+		if fp.maxCount >= fp.ltx.MaxCount {
+			if fp.ltx.AfterContext == 0 {
+				return io.EOF // Stop processing
+			}
+			fp.maxReached = true
+		}
+	}
+
+	// Reset after context
+	if fp.ltx.AfterContext > 0 {
+		fp.afterCount = fp.ltx.AfterContext
+	}
+
+	return nil
+}
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