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>

6 files changed, 1999 insertions, 0 deletions

diff --git a/internal/io/journal/filter.go b/internal/io/journal/filter.go
new file mode 100644
index 0000000..680c183
--- /dev/null
+++ b/internal/io/journal/filter.go
@@ -0,0 +1,253 @@
+//go:build linux
+
+package journal
+
+import (
+	"bytes"
+	"context"
+
+	"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"
+)
+
+type journalSink interface {
+	Emit(context.Context, *bytes.Buffer, uint64, int, string) error
+	Full() bool
+}
+
+type processorSink struct {
+	processor line.Processor
+}
+
+func (s processorSink) Emit(_ context.Context, rawLine *bytes.Buffer, count uint64,
+	_ int, sourceID string) error {
+
+	// Per the line.Processor contract, ownership of rawLine transfers to the
+	// processor: it is responsible for recycling the buffer on every return path,
+	// success or error (the journal-path processors installed via makeProcessor
+	// — DirectLineProcessor and AggregateProcessor — recycle unconditionally
+	// before returning a write error, for example). Recycling here
+	// on error would return the same buffer to the shared pool a second time; the
+	// pool would then hand one object to two Get callers and their concurrent
+	// writes would corrupt data and race. So do not recycle rawLine here.
+	return s.processor.ProcessLine(rawLine, count, sourceID)
+}
+
+func (s processorSink) Full() bool {
+	return false
+}
+
+type journalFilter struct {
+	ltx      lcontext.LContext
+	sink     journalSink
+	re       regex.Regex
+	sourceID string
+	stats    journalStats
+
+	before    []bufferedLine
+	after     int
+	maxCount  int
+	maxHit    int
+	maxClosed bool
+}
+
+type bufferedLine struct {
+	content *bytes.Buffer
+	count   uint64
+}
+
+func newJournalFilter(ltx lcontext.LContext, sink journalSink, re regex.Regex,
+	sourceID string) *journalFilter {
+
+	return &journalFilter{
+		ltx:      ltx,
+		sink:     sink,
+		re:       re,
+		sourceID: sourceID,
+		maxCount: ltx.MaxCount,
+	}
+}
+
+func (f *journalFilter) Process(ctx context.Context, rawLine *bytes.Buffer) error {
+	f.stats.updatePosition()
+	if !f.ltx.Has() {
+		return f.processWithoutContext(ctx, rawLine)
+	}
+	return f.processWithContext(ctx, rawLine)
+}
+
+func (f *journalFilter) Close() {
+	for _, line := range f.before {
+		pool.RecycleBytesBuffer(line.content)
+	}
+	f.before = nil
+}
+
+func (f *journalFilter) processWithoutContext(ctx context.Context, rawLine *bytes.Buffer) error {
+	if !f.re.Match(rawLine.Bytes()) {
+		f.stats.updateLineNotMatched()
+		f.stats.updateLineNotTransmitted()
+		pool.RecycleBytesBuffer(rawLine)
+		return nil
+	}
+
+	f.stats.updateLineMatched()
+	if f.sink.Full() {
+		f.stats.updateLineNotTransmitted()
+		pool.RecycleBytesBuffer(rawLine)
+		return nil
+	}
+	f.stats.updateLineTransmitted()
+	return f.sink.Emit(ctx, rawLine, f.stats.totalLineCount(), f.stats.transmittedPerc(), f.sourceID)
+}
+
+func (f *journalFilter) processWithContext(ctx context.Context, rawLine *bytes.Buffer) error {
+	if !f.re.Match(rawLine.Bytes()) {
+		return f.processContextMiss(ctx, rawLine)
+	}
+
+	f.stats.updateLineMatched()
+	if f.maxClosed {
+		pool.RecycleBytesBuffer(rawLine)
+		return errStopReading
+	}
+
+	if err := f.emitBefore(ctx); err != nil {
+		pool.RecycleBytesBuffer(rawLine)
+		return err
+	}
+	f.stats.updateLineTransmitted()
+	if err := f.sink.Emit(ctx, rawLine, f.stats.totalLineCount(), 100, f.sourceID); err != nil {
+		return err
+	}
+
+	if f.maxCount > 0 {
+		f.maxHit++
+		if f.maxHit >= f.maxCount {
+			if f.ltx.AfterContext == 0 {
+				return errStopReading
+			}
+			f.maxClosed = true
+		}
+	}
+	if f.ltx.AfterContext > 0 {
+		f.after = f.ltx.AfterContext
+	}
+	return nil
+}
+
+func (f *journalFilter) processContextMiss(ctx context.Context, rawLine *bytes.Buffer) error {
+	f.stats.updateLineNotMatched()
+	if f.maxClosed && f.after == 0 {
+		pool.RecycleBytesBuffer(rawLine)
+		return errStopReading
+	}
+	if f.after > 0 {
+		f.after--
+		f.stats.updateLineTransmitted()
+		err := f.sink.Emit(ctx, rawLine, f.stats.totalLineCount(), 100, f.sourceID)
+		if err == nil && f.maxClosed && f.after == 0 {
+			return errStopReading
+		}
+		return err
+	}
+	if f.ltx.BeforeContext > 0 {
+		f.rememberBefore(rawLine)
+		f.stats.updateLineNotTransmitted()
+		return nil
+	}
+
+	f.stats.updateLineNotTransmitted()
+	pool.RecycleBytesBuffer(rawLine)
+	return nil
+}
+
+func (f *journalFilter) rememberBefore(rawLine *bytes.Buffer) {
+	if len(f.before) >= f.ltx.BeforeContext {
+		pool.RecycleBytesBuffer(f.before[0].content)
+		copy(f.before, f.before[1:])
+		f.before = f.before[:len(f.before)-1]
+	}
+	f.before = append(f.before, bufferedLine{
+		content: rawLine,
+		count:   f.stats.totalLineCount(),
+	})
+}
+
+func (f *journalFilter) emitBefore(ctx context.Context) error {
+	for i, line := range f.before {
+		f.stats.updateLineTransmitted()
+		if err := f.sink.Emit(ctx, line.content, line.count, 100, f.sourceID); err != nil {
+			f.discardBeforeFrom(i + 1)
+			return err
+		}
+	}
+	f.before = f.before[:0]
+	return nil
+}
+
+func (f *journalFilter) discardBeforeFrom(index int) {
+	for _, line := range f.before[index:] {
+		pool.RecycleBytesBuffer(line.content)
+	}
+	f.before = f.before[:0]
+}
+
+type journalStats struct {
+	pos           int
+	lineCount     uint64
+	matched       [100]bool
+	matchCount    uint64
+	transmitted   [100]bool
+	transmitCount int
+}
+
+func (s *journalStats) totalLineCount() uint64 {
+	return s.lineCount
+}
+
+func (s *journalStats) transmittedPerc() int {
+	return int(percentOf(float64(s.matchCount), float64(s.transmitCount)))
+}
+
+func (s *journalStats) updatePosition() {
+	s.pos = (s.pos + 1) % 100
+	s.lineCount++
+}
+
+func (s *journalStats) updateLineMatched() {
+	if !s.matched[s.pos] {
+		s.matchCount++
+		s.matched[s.pos] = true
+	}
+}
+
+func (s *journalStats) updateLineTransmitted() {
+	if !s.transmitted[s.pos] {
+		s.transmitCount++
+		s.transmitted[s.pos] = true
+	}
+}
+
+func (s *journalStats) updateLineNotMatched() {
+	if s.matched[s.pos] {
+		s.matchCount--
+		s.matched[s.pos] = false
+	}
+}
+
+func (s *journalStats) updateLineNotTransmitted() {
+	if s.transmitted[s.pos] {
+		s.transmitCount--
+		s.transmitted[s.pos] = false
+	}
+}
+
+func percentOf(total float64, value float64) float64 {
+	if total == 0 || total == value {
+		return 100
+	}
+	return value / (total / 100.0)
+}
diff --git a/internal/io/journal/reader.go b/internal/io/journal/reader.go
new file mode 100644
index 0000000..bc27dbf
--- /dev/null
+++ b/internal/io/journal/reader.go
@@ -0,0 +1,303 @@
+//go:build linux
+
+// Package journal provides a journalctl-backed file reader.
+package journal
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"os/exec"
+	"strings"
+	"syscall"
+	"time"
+
+	"github.com/mimecast/dtail/internal/io/fs"
+	"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"
+)
+
+const (
+	defaultSourceID       = "journal"
+	journalctlCommand     = "journalctl"
+	maxScannerTokenSize   = 1024 * 1024
+	processTerminateGrace = 200 * time.Millisecond
+)
+
+var errStopReading = errors.New("stop journal reading")
+
+// ErrJournalctlNotFound reports that journalctl could not be found on PATH.
+var ErrJournalctlNotFound = errors.New("journalctl not found")
+
+// Reader reads journal entries by executing journalctl.
+type Reader struct {
+	journalctlPath string
+	args           []string
+	sourceID       string
+	serverMessages chan<- string
+	follow         bool
+}
+
+var _ fs.FileReader = (*Reader)(nil)
+
+// NewReader returns a journalctl-backed file reader.
+func NewReader(args []string, sourceID string, follow bool, serverMessages chan<- string) (*Reader, error) {
+	journalctlPath, err := exec.LookPath(journalctlCommand)
+	if err != nil {
+		return nil, fmt.Errorf("%w: %w", ErrJournalctlNotFound, err)
+	}
+	if sourceID == "" {
+		sourceID = defaultSourceID
+	}
+
+	copiedArgs := append([]string(nil), args...)
+	return &Reader{
+		journalctlPath: journalctlPath,
+		args:           copiedArgs,
+		sourceID:       sourceID,
+		serverMessages: serverMessages,
+		follow:         follow,
+	}, nil
+}
+
+// StartWithProcessor reads journalctl stdout and sends matching lines to processor.
+func (r *Reader) StartWithProcessor(ctx context.Context, ltx lcontext.LContext,
+	processor line.Processor, re regex.Regex) error {
+
+	return r.runWithProcessor(ctx, ltx, processor, re)
+}
+
+// StartWithProcessorOptimized reads journalctl stdout and sends matching lines to processor.
+func (r *Reader) StartWithProcessorOptimized(ctx context.Context, ltx lcontext.LContext,
+	processor line.Processor, re regex.Regex) error {
+
+	return r.runWithProcessor(ctx, ltx, processor, re)
+}
+
+// FilePath returns a stable journalctl command description.
+func (r *Reader) FilePath() string {
+	if len(r.args) == 0 {
+		return journalctlCommand
+	}
+	return journalctlCommand + " " + strings.Join(r.args, " ")
+}
+
+// Retry reports whether journalctl should be restarted after it exits.
+func (r *Reader) Retry() bool {
+	return r.follow
+}
+
+func (r *Reader) runWithProcessor(ctx context.Context, ltx lcontext.LContext,
+	processor line.Processor, re regex.Regex) error {
+
+	sink := processorSink{processor: processor}
+
+	// In follow mode r.run blocks until journalctl is stopped, so a batching
+	// processor (the NetworkWriter, which buffers up to 64KB before
+	// sending) would hold live lines in its buffer and the client would never
+	// see interactive output. Flush after every scanned line while following so
+	// journal follow output reaches the client promptly — the same latency
+	// guarantee the file follow path gets from its per-read-chunk flush. A
+	// non-follow snapshot read keeps the batching benefit and flushes once at
+	// the end below.
+	var flushLine func() error
+	if r.follow {
+		flushLine = processor.Flush
+	}
+
+	err := r.run(ctx, ltx, sink, re, flushLine)
+	if flushErr := processor.Flush(); flushErr != nil && err == nil {
+		err = flushErr
+	}
+	return err
+}
+
+func (r *Reader) run(ctx context.Context, ltx lcontext.LContext, sink journalSink,
+	re regex.Regex, flushLine func() error) error {
+
+	cmd := r.command(ctx)
+
+	stdout, err := cmd.StdoutPipe()
+	if err != nil {
+		return fmt.Errorf("open journalctl stdout: %w", err)
+	}
+	stderr, err := cmd.StderrPipe()
+	if err != nil {
+		return fmt.Errorf("open journalctl stderr: %w", err)
+	}
+	if err := cmd.Start(); err != nil {
+		return fmt.Errorf("start journalctl: %w", err)
+	}
+
+	stderrDone := make(chan error, 1)
+	go func() {
+		stderrDone <- r.forwardStderr(ctx, stderr)
+	}()
+
+	filter := newJournalFilter(ltx, sink, re, r.sourceID)
+	scanErr := r.scanStdout(ctx, stdout, filter, flushLine)
+	waitErr := waitForJournalctl(cmd, scanErr != nil)
+	stderrErr := <-stderrDone
+	filter.Close()
+
+	if ctx.Err() != nil || errors.Is(scanErr, errStopReading) {
+		return nil
+	}
+	if scanErr != nil {
+		return scanErr
+	}
+	if stderrErr != nil {
+		return stderrErr
+	}
+	if waitErr != nil {
+		return fmt.Errorf("journalctl failed: %w", waitErr)
+	}
+	return nil
+}
+
+func (r *Reader) command(ctx context.Context) *exec.Cmd {
+	args := r.commandArgs()
+	cmd := exec.CommandContext(ctx, r.journalctlPath, args...)
+	cmd.Cancel = func() error {
+		terminateProcess(cmd.Process)
+		return nil
+	}
+	cmd.WaitDelay = processTerminateGrace
+	return cmd
+}
+
+func (r *Reader) commandArgs() []string {
+	args := append([]string(nil), r.args...)
+	if r.follow {
+		args = append(args, "-f", "-n", "0")
+	}
+	return args
+}
+
+func terminateProcess(process *os.Process) {
+	if process == nil {
+		return
+	}
+	_ = process.Signal(syscall.SIGTERM)
+}
+
+func killProcess(process *os.Process) {
+	if process == nil {
+		return
+	}
+	_ = process.Kill()
+}
+
+func waitForJournalctl(cmd *exec.Cmd, earlyStop bool) error {
+	if !earlyStop {
+		return cmd.Wait()
+	}
+
+	terminateProcess(cmd.Process)
+
+	waitDone := make(chan error, 1)
+	go func() {
+		waitDone <- cmd.Wait()
+	}()
+
+	timer := time.NewTimer(processTerminateGrace)
+	defer timer.Stop()
+
+	select {
+	case err := <-waitDone:
+		return err
+	case <-timer.C:
+		killProcess(cmd.Process)
+		return <-waitDone
+	}
+}
+
+func (r *Reader) scanStdout(ctx context.Context, stdout io.Reader, filter *journalFilter,
+	flushLine func() error) error {
+
+	scanner := bufio.NewScanner(stdout)
+	bufPtr := pool.GetScannerBuffer()
+	defer pool.PutScannerBuffer(bufPtr)
+
+	scanner.Buffer(*bufPtr, maxScannerTokenSize)
+	scanner.Split(scanLinesPreserveEndings)
+
+	for scanner.Scan() {
+		select {
+		case <-ctx.Done():
+			return nil
+		default:
+		}
+
+		lineBuf := pool.BytesBuffer.Get().(*bytes.Buffer)
+		lineBuf.Write(scanner.Bytes())
+		if err := filter.Process(ctx, lineBuf); err != nil {
+			return err
+		}
+
+		// In follow mode, push any buffered line straight to the client so
+		// batching never delays live output. flushLine is nil for non-follow
+		// reads (they batch and flush once at the end) and for the immediate
+		// channel sink.
+		if flushLine != nil {
+			if err := flushLine(); err != nil {
+				return err
+			}
+		}
+	}
+	if err := scanner.Err(); err != nil {
+		return fmt.Errorf("scan journalctl stdout: %w", err)
+	}
+	return nil
+}
+
+func (r *Reader) forwardStderr(ctx context.Context, stderr io.Reader) error {
+	scanner := bufio.NewScanner(stderr)
+	bufPtr := pool.GetScannerBuffer()
+	defer pool.PutScannerBuffer(bufPtr)
+
+	scanner.Buffer(*bufPtr, maxScannerTokenSize)
+	for scanner.Scan() {
+		if !r.sendServerMessage(ctx, fmt.Sprintf("journalctl stderr: %s\n", scanner.Text())) {
+			return nil
+		}
+	}
+	if err := scanner.Err(); err != nil {
+		if errors.Is(err, os.ErrClosed) {
+			return nil
+		}
+		return fmt.Errorf("scan journalctl stderr: %w", err)
+	}
+	return nil
+}
+
+func (r *Reader) sendServerMessage(ctx context.Context, message string) bool {
+	if r.serverMessages == nil {
+		return true
+	}
+	select {
+	case r.serverMessages <- message:
+		return true
+	case <-ctx.Done():
+		return false
+	}
+}
+
+func scanLinesPreserveEndings(data []byte, atEOF bool) (int, []byte, error) {
+	if atEOF && len(data) == 0 {
+		return 0, nil, nil
+	}
+	if i := bytes.IndexByte(data, '\n'); i >= 0 {
+		return i + 1, data[:i+1], nil
+	}
+	if atEOF {
+		return len(data), nil, nil
+	}
+	return 0, nil, nil
+}
diff --git a/internal/io/journal/reader_test.go b/internal/io/journal/reader_test.go
new file mode 100644
index 0000000..a2d084f
--- /dev/null
+++ b/internal/io/journal/reader_test.go
@@ -0,0 +1,754 @@
+//go:build linux
+
+package journal
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"os"
+	"os/exec"
+	"reflect"
+	"strconv"
+	"strings"
+	"sync"
+	"syscall"
+	"testing"
+	"time"
+
+	journaltest "github.com/mimecast/dtail/internal/io/journal/testhelper"
+	"github.com/mimecast/dtail/internal/io/pool"
+	"github.com/mimecast/dtail/internal/lcontext"
+	"github.com/mimecast/dtail/internal/regex"
+)
+
+type captureProcessor struct {
+	lines []string
+}
+
+func (p *captureProcessor) ProcessLine(lineContent *bytes.Buffer, _ uint64, _ string) error {
+	p.lines = append(p.lines, lineContent.String())
+	pool.RecycleBytesBuffer(lineContent)
+	return nil
+}
+
+func (p *captureProcessor) Flush() error {
+	return nil
+}
+
+func (p *captureProcessor) Close() error {
+	return nil
+}
+
+type errorProcessor struct {
+	err error
+}
+
+func (p errorProcessor) ProcessLine(lineContent *bytes.Buffer, _ uint64, _ string) error {
+	pool.RecycleBytesBuffer(lineContent)
+	return p.err
+}
+
+func (p errorProcessor) Flush() error {
+	return nil
+}
+
+func (p errorProcessor) Close() error {
+	return nil
+}
+
+// nonRecyclingErrorProcessor returns an error without recycling the buffer, so a
+// test can observe whether processorSink.Emit wrongly recycles a buffer it does
+// not own.
+type nonRecyclingErrorProcessor struct {
+	err error
+}
+
+func (p nonRecyclingErrorProcessor) ProcessLine(_ *bytes.Buffer, _ uint64, _ string) error {
+	return p.err
+}
+
+func (p nonRecyclingErrorProcessor) Flush() error { return nil }
+
+func (p nonRecyclingErrorProcessor) Close() error { return nil }
+
+// TestProcessorSinkEmitDoesNotRecycleOnError is the regression guard for the
+// bt0 data race. The line.Processor contract transfers buffer ownership to the
+// processor, which recycles it on every return path (the journal-path processors
+// DirectLineProcessor and AggregateProcessor recycle unconditionally before
+// returning a write error). If
+// processorSink.Emit also recycled the buffer on error, the same buffer would be
+// returned to the shared pool twice; the pool would then hand one object to two
+// Get callers whose concurrent writes race and corrupt data. Emit must therefore
+// leave the buffer untouched on error. RecycleBytesBuffer calls buf.Reset(), so
+// a stray recycle would clear the payload — assert it survives.
+func TestProcessorSinkEmitDoesNotRecycleOnError(t *testing.T) {
+	buf := pool.BytesBuffer.Get().(*bytes.Buffer)
+	buf.Reset()
+	buf.WriteString("payload")
+
+	sinkErr := errors.New("processor stopped")
+	sink := processorSink{processor: nonRecyclingErrorProcessor{err: sinkErr}}
+
+	err := sink.Emit(context.Background(), buf, 1, 100, "journal-id")
+	if !errors.Is(err, sinkErr) {
+		t.Fatalf("Emit error = %v, want %v", err, sinkErr)
+	}
+	if got := buf.String(); got != "payload" {
+		t.Fatalf("processorSink.Emit recycled a buffer it does not own (double-recycle regression): buf=%q", got)
+	}
+
+	pool.RecycleBytesBuffer(buf)
+}
+
+func TestNewReaderFailsWhenJournalctlIsMissing(t *testing.T) {
+	t.Setenv("PATH", t.TempDir())
+
+	reader, err := NewReader(nil, "journal", false, nil)
+	if err == nil {
+		t.Fatal("expected missing journalctl error")
+	}
+	if reader != nil {
+		t.Fatalf("expected nil reader, got %#v", reader)
+	}
+	if !errors.Is(err, ErrJournalctlNotFound) {
+		t.Fatalf("missing journalctl error = %v, want ErrJournalctlNotFound", err)
+	}
+	if !errors.Is(err, exec.ErrNotFound) {
+		t.Fatalf("missing journalctl error = %v, want exec.ErrNotFound", err)
+	}
+}
+
+func TestStartReadsJournalctlOutputWithoutFollowFlags(t *testing.T) {
+	mock := journaltest.InstallMock(t, journaltest.Scenario{
+		Default: journaltest.Invocation{
+			Lines: []string{"alpha", "beta"},
+		},
+	})
+
+	reader, err := NewReader([]string{"-u", "ssh.service"}, "journal-id", false, make(chan string, 1))
+	if err != nil {
+		t.Fatalf("new reader: %v", err)
+	}
+
+	processor := &captureProcessor{}
+	if err := reader.StartWithProcessorOptimized(context.Background(), lcontext.LContext{},
+		processor, regex.NewNoop()); err != nil {
+		t.Fatalf("start reader: %v", err)
+	}
+
+	want := []string{"alpha\n", "beta\n"}
+	if !reflect.DeepEqual(processor.lines, want) {
+		t.Fatalf("unexpected lines: got=%v want=%v", processor.lines, want)
+	}
+
+	args := mock.Args(t)
+	if strings.Contains(args, "-f") || strings.Contains(args, "-n 0") {
+		t.Fatalf("non-follow reader passed follow flags: %q", args)
+	}
+	if strings.TrimSpace(args) != "-u ssh.service" {
+		t.Fatalf("unexpected journalctl args: %q", args)
+	}
+	if reader.Retry() {
+		t.Fatal("non-follow reader should not retry")
+	}
+}
+
+func TestStartFollowReadsLinesInOrder(t *testing.T) {
+	journaltest.InstallMock(t, journaltest.Scenario{
+		Default: journaltest.Invocation{
+			FollowLines:    []string{"alpha", "beta", "gamma"},
+			InterLineDelay: 5 * time.Millisecond,
+		},
+	})
+
+	reader, err := NewReader([]string{"-u", "ssh.service"}, "journal-id", true, make(chan string, 8))
+	if err != nil {
+		t.Fatalf("new reader: %v", err)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	processor := &flushCountingProcessor{}
+	done := make(chan error, 1)
+	go func() {
+		done <- reader.StartWithProcessorOptimized(ctx, lcontext.LContext{}, processor, regex.NewNoop())
+	}()
+
+	// Poll until all three follow lines have arrived (the reader appends from its
+	// goroutine while following).
+	deadline := time.After(2 * time.Second)
+	var got []string
+	for {
+		got = processor.snapshot()
+		if len(got) >= 3 {
+			break
+		}
+		select {
+		case <-deadline:
+			t.Fatalf("follow reader did not deliver 3 lines: got=%v", got)
+		case <-time.After(2 * time.Millisecond):
+		}
+	}
+
+	cancel()
+	select {
+	case err := <-done:
+		if err != nil && !errors.Is(err, context.Canceled) {
+			t.Fatalf("follow reader returned error after cancel: %v", err)
+		}
+	case <-time.After(500 * time.Millisecond):
+		t.Fatal("follow reader did not stop promptly after cancellation")
+	}
+
+	want := []string{"alpha\n", "beta\n", "gamma\n"}
+	if !reflect.DeepEqual(got[:3], want) {
+		t.Fatalf("unexpected follow lines: got=%v want=%v", got, want)
+	}
+}
+
+func TestStartFollowPassesFollowFlagsAndTerminatesOnCancel(t *testing.T) {
+	mock := journaltest.InstallMock(t, journaltest.Scenario{
+		Default: journaltest.Invocation{
+			Lines: []string{"ready"},
+		},
+	})
+
+	reader, err := NewReader([]string{"-u", "ssh.service"}, "journal-id", true, make(chan string, 1))
+	if err != nil {
+		t.Fatalf("new reader: %v", err)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	processor := &flushCountingProcessor{}
+	done := make(chan error, 1)
+	go func() {
+		done <- reader.StartWithProcessorOptimized(ctx, lcontext.LContext{}, processor, regex.NewNoop())
+	}()
+
+	// Wait for the first follow line to arrive before probing the process.
+	deadline := time.After(2 * time.Second)
+	for {
+		got := processor.snapshot()
+		if len(got) >= 1 {
+			if got[0] != "ready\n" {
+				t.Fatalf("unexpected first line: %q", got[0])
+			}
+			break
+		}
+		select {
+		case <-deadline:
+			t.Fatal("follow reader did not deliver the first line")
+		case <-time.After(2 * time.Millisecond):
+		}
+	}
+
+	pid := mockPID(t, mock)
+	if !processExists(pid) {
+		t.Fatalf("fake journalctl process %d does not exist before cancel", pid)
+	}
+
+	started := time.Now()
+	cancel()
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatalf("follow reader returned error after cancel: %v", err)
+		}
+	case <-time.After(500 * time.Millisecond):
+		t.Fatal("follow reader did not stop promptly after cancellation")
+	}
+	if elapsed := time.Since(started); elapsed > 500*time.Millisecond {
+		t.Fatalf("follow reader returned too slowly after cancel: %s", elapsed)
+	}
+
+	args := mock.Args(t)
+	if !strings.Contains(args, "-f -n 0") {
+		t.Fatalf("follow reader did not pass follow flags: %q", args)
+	}
+	if !mock.Terminated(t) {
+		t.Fatal("journalctl did not observe SIGTERM")
+	}
+	if processExists(pid) {
+		t.Fatalf("fake journalctl process %d still exists after reader returned", pid)
+	}
+	if !reader.Retry() {
+		t.Fatal("follow reader should retry")
+	}
+}
+
+func TestStartSurfacesStderrAsServerMessages(t *testing.T) {
+	journaltest.InstallMock(t, journaltest.Scenario{
+		Default: journaltest.Invocation{
+			Lines:  []string{"alpha"},
+			Stderr: []string{"journal warning"},
+		},
+	})
+
+	serverMessages := make(chan string, 1)
+	reader, err := NewReader(nil, "journal-id", false, serverMessages)
+	if err != nil {
+		t.Fatalf("new reader: %v", err)
+	}
+
+	if err := reader.StartWithProcessorOptimized(context.Background(), lcontext.LContext{},
+		&captureProcessor{}, regex.NewNoop()); err != nil {
+		t.Fatalf("start reader: %v", err)
+	}
+
+	select {
+	case message := <-serverMessages:
+		if message != "journalctl stderr: journal warning\n" {
+			t.Fatalf("unexpected server message: %q", message)
+		}
+	default:
+		t.Fatal("expected stderr server message")
+	}
+}
+
+func TestStartReturnsExitErrorAndForwardsStderrOnNonZeroExit(t *testing.T) {
+	journaltest.InstallMock(t, journaltest.Scenario{
+		Default: journaltest.Invocation{
+			Lines:    []string{"before failure"},
+			Stderr:   []string{"boom"},
+			ExitCode: 17,
+		},
+	})
+
+	serverMessages := make(chan string, 2)
+	reader, err := NewReader(nil, "journal-id", false, serverMessages)
+	if err != nil {
+		t.Fatalf("new reader: %v", err)
+	}
+
+	processor := &captureProcessor{}
+	err = reader.StartWithProcessorOptimized(context.Background(), lcontext.LContext{},
+		processor, regex.NewNoop())
+	if err == nil {
+		t.Fatal("expected non-zero journalctl exit error")
+	}
+	if !strings.Contains(err.Error(), "journalctl failed") {
+		t.Fatalf("unexpected non-zero exit error: %v", err)
+	}
+	var exitErr *exec.ExitError
+	if !errors.As(err, &exitErr) {
+		t.Fatalf("error = %v, want exec.ExitError", err)
+	}
+	if got := exitErr.ExitCode(); got != 17 {
+		t.Fatalf("exit code = %d, want 17", got)
+	}
+	if reader.Retry() {
+		t.Fatal("non-follow reader should not retry after non-zero exit")
+	}
+
+	if want := []string{"before failure\n"}; !reflect.DeepEqual(processor.lines, want) {
+		t.Fatalf("unexpected lines before failure: got=%v want=%v", processor.lines, want)
+	}
+
+	select {
+	case message := <-serverMessages:
+		if message != "journalctl stderr: boom\n" {
+			t.Fatalf("unexpected server message: %q", message)
+		}
+	default:
+		t.Fatal("expected stderr server message")
+	}
+}
+
+func TestStartReadsLongJournalLine(t *testing.T) {
+	const longLineLength = 70 * 1024
+
+	journaltest.InstallMock(t, journaltest.Scenario{
+		Default: journaltest.Invocation{
+			LongLineLength: longLineLength,
+		},
+	})
+
+	reader, err := NewReader(nil, "journal-id", false, nil)
+	if err != nil {
+		t.Fatalf("new reader: %v", err)
+	}
+
+	processor := &captureProcessor{}
+	if err := reader.StartWithProcessorOptimized(context.Background(), lcontext.LContext{},
+		processor, regex.NewNoop()); err != nil {
+		t.Fatalf("start reader: %v", err)
+	}
+
+	want := []string{strings.Repeat("x", longLineLength) + "\n"}
+	if !reflect.DeepEqual(processor.lines, want) {
+		gotLen := 0
+		if len(processor.lines) > 0 {
+			gotLen = len(processor.lines[0])
+		}
+		t.Fatalf("unexpected long line: got len=%d want len=%d", gotLen, len(want[0]))
+	}
+}
+
+func TestStartDropsPartialLineAtShutdown(t *testing.T) {
+	journaltest.InstallMock(t, journaltest.Scenario{
+		Default: journaltest.Invocation{
+			PartialLine: "unterminated",
+		},
+	})
+
+	reader, err := NewReader(nil, "journal-id", false, nil)
+	if err != nil {
+		t.Fatalf("new reader: %v", err)
+	}
+
+	processor := &captureProcessor{}
+	if err := reader.StartWithProcessorOptimized(context.Background(), lcontext.LContext{},
+		processor, regex.NewNoop()); err != nil {
+		t.Fatalf("start reader: %v", err)
+	}
+	if len(processor.lines) != 0 {
+		t.Fatalf("partial line without trailing newline was emitted: %v", processor.lines)
+	}
+}
+
+func TestStartPreservesUTF8Lines(t *testing.T) {
+	want := []string{"żółć 🚀\n", "東京 café\n"}
+	journaltest.InstallMock(t, journaltest.Scenario{
+		Default: journaltest.Invocation{
+			Lines: []string{strings.TrimSuffix(want[0], "\n"), strings.TrimSuffix(want[1], "\n")},
+		},
+	})
+
+	reader, err := NewReader(nil, "journal-id", false, nil)
+	if err != nil {
+		t.Fatalf("new reader: %v", err)
+	}
+
+	processor := &captureProcessor{}
+	if err := reader.StartWithProcessorOptimized(context.Background(), lcontext.LContext{},
+		processor, regex.NewNoop()); err != nil {
+		t.Fatalf("start reader: %v", err)
+	}
+
+	if !reflect.DeepEqual(processor.lines, want) {
+		t.Fatalf("unexpected UTF-8 lines: got=%v want=%v", processor.lines, want)
+	}
+}
+
+func TestConcurrentReadersDifferentUnitsDoNotInterfere(t *testing.T) {
+	journaltest.InstallMock(t, journaltest.Scenario{
+		Units: map[string]journaltest.Invocation{
+			"alpha.service": {
+				Lines: []string{"alpha-1", "alpha-2"},
+			},
+			"beta.service": {
+				Lines: []string{"beta-1", "beta-2"},
+			},
+		},
+	})
+
+	type result struct {
+		lines []string
+		err   error
+	}
+
+	runReader := func(unit string) <-chan result {
+		resultCh := make(chan result, 1)
+		go func() {
+			reader, err := NewReader([]string{"-u", unit}, unit, false, nil)
+			if err != nil {
+				resultCh <- result{err: err}
+				return
+			}
+			processor := &captureProcessor{}
+			err = reader.StartWithProcessorOptimized(context.Background(), lcontext.LContext{},
+				processor, regex.NewNoop())
+			resultCh <- result{lines: processor.lines, err: err}
+		}()
+		return resultCh
+	}
+
+	alphaCh := runReader("alpha.service")
+	betaCh := runReader("beta.service")
+
+	alpha := <-alphaCh
+	beta := <-betaCh
+
+	if alpha.err != nil {
+		t.Fatalf("alpha reader: %v", alpha.err)
+	}
+	if beta.err != nil {
+		t.Fatalf("beta reader: %v", beta.err)
+	}
+	if want := []string{"alpha-1\n", "alpha-2\n"}; !reflect.DeepEqual(alpha.lines, want) {
+		t.Fatalf("alpha lines: got=%v want=%v", alpha.lines, want)
+	}
+	if want := []string{"beta-1\n", "beta-2\n"}; !reflect.DeepEqual(beta.lines, want) {
+		t.Fatalf("beta lines: got=%v want=%v", beta.lines, want)
+	}
+}
+
+func TestStartWithProcessorOptimizedAppliesRegexAndLocalContext(t *testing.T) {
+	journaltest.InstallMock(t, journaltest.Scenario{
+		Default: journaltest.Invocation{
+			Lines: []string{"before", "match", "context", "skip"},
+		},
+	})
+
+	reader, err := NewReader(nil, "journal-id", false, nil)
+	if err != nil {
+		t.Fatalf("new reader: %v", err)
+	}
+	re, err := regex.New("match", regex.Default)
+	if err != nil {
+		t.Fatalf("new regex: %v", err)
+	}
+	processor := &captureProcessor{}
+
+	err = reader.StartWithProcessorOptimized(
+		context.Background(),
+		lcontext.LContext{BeforeContext: 1, AfterContext: 1, MaxCount: 1},
+		processor,
+		re,
+	)
+	if err != nil && !errors.Is(err, context.Canceled) {
+		t.Fatalf("start optimized reader: %v", err)
+	}
+
+	want := []string{"before\n", "match\n", "context\n"}
+	if !reflect.DeepEqual(processor.lines, want) {
+		t.Fatalf("unexpected processed lines: got=%v want=%v", processor.lines, want)
+	}
+}
+
+// flushCountingProcessor records processed lines and how many times Flush was
+// invoked, guarded by a mutex because the reader drives it from a goroutine.
+type flushCountingProcessor struct {
+	mu         sync.Mutex
+	lines      []string
+	flushCount int
+}
+
+func (p *flushCountingProcessor) ProcessLine(lineContent *bytes.Buffer, _ uint64, _ string) error {
+	p.mu.Lock()
+	p.lines = append(p.lines, lineContent.String())
+	p.mu.Unlock()
+	pool.RecycleBytesBuffer(lineContent)
+	return nil
+}
+
+func (p *flushCountingProcessor) Flush() error {
+	p.mu.Lock()
+	p.flushCount++
+	p.mu.Unlock()
+	return nil
+}
+
+func (p *flushCountingProcessor) Close() error { return nil }
+
+func (p *flushCountingProcessor) counts() (lines, flushes int) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	return len(p.lines), p.flu