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, 462 insertions, 51 deletions

diff --git a/internal/server/handlers/readcommand.go b/internal/server/handlers/readcommand.go
index 44ba9e4..59254e8 100644
--- a/internal/server/handlers/readcommand.go
+++ b/internal/server/handlers/readcommand.go
@@ -1,58 +1,87 @@
 package handlers
 
 import (
+	"bytes"
 	"context"
+	"io"
 	"os"
 	"path/filepath"
 	"strings"
 	"sync"
 	"time"
 
+	"github.com/mimecast/dtail/internal/config"
+	"github.com/mimecast/dtail/internal/ctxutil"
 	"github.com/mimecast/dtail/internal/io/dlog"
 	"github.com/mimecast/dtail/internal/io/fs"
-	"github.com/mimecast/dtail/internal/io/line"
+	"github.com/mimecast/dtail/internal/io/journal"
 	"github.com/mimecast/dtail/internal/lcontext"
+	"github.com/mimecast/dtail/internal/mapr/server"
 	"github.com/mimecast/dtail/internal/omode"
 	"github.com/mimecast/dtail/internal/regex"
 )
 
 type readCommand struct {
-	server *ServerHandler
-	mode   omode.Mode
+	server              readCommandServer
+	mode                omode.Mode
+	generation          uint64
+	shutdownCoordinator *shutdownCoordinator
 }
 
-func newReadCommand(server *ServerHandler, mode omode.Mode) *readCommand {
+type readStrategy func(context.Context, lcontext.LContext, fs.FileReader, regex.Regex) error
+
+type readProcessor interface {
+	ProcessLine(*bytes.Buffer, uint64, string) error
+	Flush() error
+	Close() error
+}
+
+func newReadCommand(server readCommandServer, mode omode.Mode) *readCommand {
+	// cat/grep reads are one-shot: their input is exhausted once every file
+	// has been read to EOF. tail follows its files indefinitely, so its input
+	// never exhausts and must not finish a output aggregate.
+	oneShotInput := mode == omode.CatClient || mode == omode.GrepClient
 	return &readCommand{
-		server: server,
-		mode:   mode,
+		server:              server,
+		mode:                mode,
+		shutdownCoordinator: newShutdownCoordinator(server, oneShotInput),
 	}
 }
 
 func (r *readCommand) Start(ctx context.Context, ltx lcontext.LContext,
 	argc int, args []string, retries int) {
+	r.generation = sessionGenerationFromContext(ctx)
 
 	re := regex.NewNoop()
 	if argc >= 4 {
 		deserializedRegex, err := regex.Deserialize(strings.Join(args[2:], " "))
 		if err != nil {
-			r.server.sendln(r.server.serverMessages, dlog.Server.Error(r.server.user,
+			r.sendServerMessage(ctx, dlog.Server.Error(r.server.LogContext(),
 				"Unable to parse command", err))
 			return
 		}
 		re = deserializedRegex
 	}
 	if argc < 3 {
-		r.server.sendln(r.server.serverMessages, dlog.Server.Warn(r.server.user,
+		r.sendServerMessage(ctx, dlog.Server.Warn(r.server.LogContext(),
 			"Unable to parse command", args, argc))
 		return
 	}
 
 	// In serverless mode, can also read data from pipe
 	// e.g.: grep foo bar.log | dmap 'from STATS select ...'
-	if r.isInputFromPipe() {
+	// Only read from pipe if no file argument is provided
+	isPipe := r.isInputFromPipe() && (argc < 2 || args[1] == "" || args[1] == "-")
+
+	if isPipe {
 		dlog.Server.Debug("Reading data from stdin pipe")
-		// Empty file path and globID "-" represents reading from the stdin pipe.
-		r.read(ctx, ltx, "", "-", re)
+		r.readPipe(ctx, ltx, re)
+		return
+	}
+
+	if fs.IsJournalSpec(args[1]) {
+		dlog.Server.Debug("Reading data from journal")
+		r.readJournal(ctx, ltx, args[1], re, retries)
 		return
 	}
 
@@ -60,38 +89,82 @@ func (r *readCommand) Start(ctx context.Context, ltx lcontext.LContext,
 	r.readGlob(ctx, ltx, args[1], re, retries)
 }
 
+// readPipe reads the single stdin-pipe input in serverless mode (e.g.
+// `grep foo bar.log | dmap 'from STATS select ...'`). Unlike file/glob/journal
+// reads it does NOT go through readFiles/readFileIfPermissions, so it must
+// reproduce their finalization here: the pipe is accounted as one pending input
+// and onFileProcessed is invoked once it drains. That is what drives the
+// shutdown coordinator's FinishInput signal to a output aggregate — without it a
+// serverless dmap reading from a pipe would block in Aggregate.Start on its
+// inputFinished signal forever (the client keeps re-gathering interim results
+// and never terminates). Previously the pipe fed a regular Aggregate that
+// finalized when its input line channel was closed; the output aggregate has no
+// such channel-close, so the pending-file/onFileProcessed path is the
+// equivalent "input exhausted" signal. The pending accounting also keeps the
+// counter balanced: onFileProcessed unconditionally decrements pendingFiles, so
+// it must be paired with an AddPendingFiles here.
+func (r *readCommand) readPipe(ctx context.Context, ltx lcontext.LContext, re regex.Regex) {
+	r.server.AddPendingFiles(1)
+	defer r.shutdownCoordinator.onFileProcessed("-")
+	// Empty file path and globID "-" represents reading from the stdin pipe.
+	r.read(ctx, ltx, "", nil, "-", re)
+}
+
+func (r *readCommand) readJournal(ctx context.Context, ltx lcontext.LContext,
+	spec string, re regex.Regex, _ int) {
+
+	r.readFiles(ctx, ltx, []string{spec}, spec, re, r.server.ReadGlobRetryInterval())
+}
+
 func (r *readCommand) readGlob(ctx context.Context, ltx lcontext.LContext,
 	glob string, re regex.Regex, retries int) {
 
-	retryInterval := time.Second * 5
+	retryInterval := r.server.ReadGlobRetryInterval()
 	glob = filepath.Clean(glob)
 
 	for retryCount := 0; retryCount < retries; retryCount++ {
 		paths, err := filepath.Glob(glob)
 		if err != nil {
-			dlog.Server.Warn(r.server.user, glob, err)
-			time.Sleep(retryInterval)
+			dlog.Server.Warn(r.server.LogContext(), glob, err)
+			if !ctxutil.Sleep(ctx, retryInterval) {
+				return
+			}
 			continue
 		}
 
 		if numPaths := len(paths); numPaths == 0 {
-			dlog.Server.Error(r.server.user, "No such file(s) to read", glob)
-			r.server.sendln(r.server.serverMessages, dlog.Server.Warn(r.server.user,
+			dlog.Server.Error(r.server.LogContext(), "No such file(s) to read", glob)
+			r.sendServerMessage(ctx, dlog.Server.Warn(r.server.LogContext(),
 				"Unable to read file(s), check server logs"))
 			select {
 			case <-ctx.Done():
 				return
 			default:
 			}
-			time.Sleep(retryInterval)
+			if !ctxutil.Sleep(ctx, retryInterval) {
+				return
+			}
 			continue
 		}
 
+		// Cap the number of paths to prevent an authenticated user with a broad
+		// read permission from spawning an unbounded number of goroutines and
+		// exhausting server memory. Excess paths are dropped with a warning so
+		// the partial result is still delivered rather than failing entirely.
+		if cap := r.server.MaxGlobTargets(); len(paths) > cap {
+			dlog.Server.Warn(r.server.LogContext(), "Glob expansion exceeded cap, truncating",
+				"glob", glob, "matched", len(paths), "cap", cap)
+			r.sendServerMessage(ctx, dlog.Server.Warn(r.server.LogContext(),
+				"Glob expansion exceeded server limit, only first targets served",
+				"limit", cap, "matched", len(paths)))
+			paths = paths[:cap]
+		}
+
 		r.readFiles(ctx, ltx, paths, glob, re, retryInterval)
 		return
 	}
 
-	r.server.sendln(r.server.serverMessages, dlog.Server.Warn(r.server.user,
+	r.sendServerMessage(ctx, dlog.Server.Warn(r.server.LogContext(),
 		"Giving up to read file(s)"))
 	return
 }
@@ -99,81 +172,255 @@ func (r *readCommand) readGlob(ctx context.Context, ltx lcontext.LContext,
 func (r *readCommand) readFiles(ctx context.Context, ltx lcontext.LContext,
 	paths []string, glob string, re regex.Regex, retryInterval time.Duration) {
 
+	dlog.Server.Info(r.server.LogContext(), "Processing files", "count", len(paths), "glob", glob)
+
+	// Track pending files for this batch
+	totalPending := r.server.AddPendingFiles(int32(len(paths)))
+	dlog.Server.Info(r.server.LogContext(), "Added pending files", "count", len(paths), "totalPending", totalPending)
+
 	var wg sync.WaitGroup
 	wg.Add(len(paths))
 	for _, path := range paths {
 		go r.readFileIfPermissions(ctx, ltx, &wg, path, glob, re)
 	}
 	wg.Wait()
+
+	dlog.Server.Info(r.server.LogContext(), "All files processed", "count", len(paths))
+
+	select {
+	case <-ctx.Done():
+		return
+	default:
+	}
+
+	// In output mode, signal EOF once all pending file work is drained.
+	// Active command count may still include side-effect commands (for example AUTHKEY),
+	// so relying on "active == 1" can skip EOF signaling and lead to dropped output.
+	//
+	// Output is now the only runtime path, so the former config gating on
+	// the former config gating has been removed. This EOF handshake runs for every
+	// cat/grep/tail read.
+	//
+	// The guard is the mode check rather than Aggregate() == nil:
+	// SERVER-MODE dmap also enables output mode (readWithProcessor ->
+	// ensureOutputEnabled) even though its lines feed the Aggregate
+	// rather than the output channel writer, and it RELIES on this epilogue's
+	// SignalOutputEOF to disable output mode on the session output goroutine so the
+	// client receives EOF and the session terminates. Excluding the
+	// output-aggregate case here (as Aggregate() == nil would) hangs
+	// server-mode dmap: output mode stays enabled and the client waits forever for
+	// output that never ends. dmap uses map mode (not cat/grep/tail) so it is
+	// already excluded from this direct-output epilogue.
+	//
+	// For serverless dmap the epilogue is likewise not reached (map mode): the
+	// output aggregate is finalized through FinishInput in the shutdown
+	// coordinator, independently of this direct-output handshake.
+	if r.mode == omode.CatClient || r.mode == omode.GrepClient || r.mode == omode.TailClient {
+		if r.server.DirectOutputActive() && r.server.HasOutputEOF() {
+			// Capture the handshake epoch BEFORE the pending check. A command
+			// that joins the session afterwards bumps the epoch (readFiles
+			// increments the pending count before its per-file work enables
+			// output mode), so a joiner invisible to the pending==0 check below
+			// is guaranteed to advance the epoch after this capture — turning
+			// our SignalOutputEOF into a no-op instead of cutting off the
+			// joiner's output mid-batch. The window between the checks below
+			// and the signal is wide (FlushOutput may block for seconds),
+			// which is exactly what this guard covers.
+			//
+			// Note the never-signaling-joiner class: a output-aggregate (dmap)
+			// command joining the live session also bumps the epoch via its
+			// enable, but map mode is excluded from this epilogue and never
+			// signals EOF itself. Our signal is then dropped and the ack wait
+			// below runs into its bounded timeout — a spurious warning, but no
+			// data loss: our data was flushed above and session shutdown
+			// flushes the rest (see outputManager.signalEOF).
+			epoch := r.server.OutputEpoch()
+
+			pending, active := r.server.PendingAndActive()
+			shouldSignalEOF := pending == 0
+			if !shouldSignalEOF {
+				dlog.Server.Trace(r.server.LogContext(), "Skipping output EOF signal for non-final command",
+					"pending", pending, "active", active)
+				return
+			}
+
+			dlog.Server.Debug(r.server.LogContext(), "Output mode: flushing data before EOF signal")
+
+			// Ensure all output data is flushed before signaling EOF.
+			r.server.FlushOutput()
+
+			// Signal EOF by closing the channel, but only once — and only if
+			// no newer batch joined since the epoch capture above.
+			r.server.SignalOutputEOF(epoch)
+
+			// Wait for an explicit reader acknowledgement instead of timing guesses.
+			if !r.server.Serverless() {
+				timeout := r.server.OutputEOFAckTimeout()
+				if r.server.WaitForOutputEOFAck(timeout) {
+					// The wait is also released when enable() hands the
+					// handshake over to a new batch (stale refresh), not only
+					// by a reader ack — the log wording covers both.
+					dlog.Server.Debug(r.server.LogContext(), "Output EOF handshake released (reader ack or handover)")
+					// Allow transport buffers to flush after acknowledgement.
+					if !ctxutil.Sleep(ctx, r.server.ShutdownSerializeWait()) {
+						return
+					}
+				} else {
+					dlog.Server.Warn(
+						r.server.LogContext(),
+						"Timeout waiting for output EOF acknowledgement",
+						"timeout", timeout,
+						"remaining", r.server.OutputChannelLen(),
+					)
+				}
+			}
+		}
+	}
+
+	// In output mode with aggregate, we don't close the shared channel here
+	// because it will be used across multiple invocations
+	// The aggregate will handle channel closure when it's done
 }
 
 func (r *readCommand) readFileIfPermissions(ctx context.Context, ltx lcontext.LContext,
 	wg *sync.WaitGroup, path, glob string, re regex.Regex) {
 
 	defer wg.Done()
-	globID := r.makeGlobID(path, glob)
-	if !r.server.user.HasFilePermission(path, "readfiles") {
-		dlog.Server.Error(r.server.user, "No permission to read file", path, globID)
-		r.server.sendln(r.server.serverMessages, dlog.Server.Warn(r.server.user,
+	defer func() {
+		r.shutdownCoordinator.onFileProcessed(path)
+	}()
+
+	globID := r.makeGlobID(ctx, path, glob)
+	target, ok := r.server.PrepareReadTarget(path)
+	if !ok {
+		dlog.Server.Error(r.server.LogContext(), "No permission to read file", path, globID)
+		r.sendServerMessage(ctx, dlog.Server.Warn(r.server.LogContext(),
 			"Unable to read file(s), check server logs"))
 		return
 	}
-	r.read(ctx, ltx, path, globID, re)
+	r.read(ctx, ltx, path, &target, globID, re)
 }
 
 func (r *readCommand) read(ctx context.Context, ltx lcontext.LContext,
-	path, globID string, re regex.Regex) {
+	path string, target *fs.ValidatedReadTarget, globID string, re regex.Regex) {
+
+	dlog.Server.Info(r.server.LogContext(), "Start reading", path, globID)
+	r.logRegexMode(re)
 
-	dlog.Server.Info(r.server.user, "Start reading", path, globID)
 	var reader fs.FileReader
 	var limiter chan struct{}
+	serverMessages, closeServerMessages := r.newGeneratedServerMessagesChannel(ctx)
+	defer closeServerMessages()
 
 	switch r.mode {
 	case omode.GrepClient, omode.CatClient:
-		reader = fs.NewCatFile(path, globID, r.server.serverMessages)
-		limiter = r.server.catLimiter
+		if target != nil && target.Kind == fs.JournalKind {
+			journalReader, err := journal.NewReader(journalArgs(path), path, false, serverMessages)
+			if err != nil {
+				r.sendServerMessage(ctx, dlog.Server.Warn(r.server.LogContext(), "Unable to read journal", err))
+				return
+			}
+			reader = journalReader
+		} else if target != nil {
+			catFile := fs.NewValidatedCatFile(path, *target, globID, serverMessages, r.server.MaxLineLength())
+			reader = &catFile
+		} else {
+			catFile := fs.NewCatFile(path, globID, serverMessages, r.server.MaxLineLength())
+			reader = &catFile
+		}
+		limiter = r.server.CatLimiter()
 	case omode.TailClient:
 		fallthrough
 	default:
-		reader = fs.NewTailFile(path, globID, r.server.serverMessages)
-		limiter = r.server.tailLimiter
+		if target != nil && target.Kind == fs.JournalKind {
+			journalReader, err := journal.NewReader(journalArgs(path), path, true, serverMessages)
+			if err != nil {
+				r.sendServerMessage(ctx, dlog.Server.Warn(r.server.LogContext(), "Unable to read journal", err))
+				return
+			}
+			reader = journalReader
+		} else if target != nil {
+			tailFile := fs.NewValidatedTailFile(path, *target, globID, serverMessages, r.server.MaxLineLength())
+			reader = &tailFile
+		} else {
+			tailFile := fs.NewTailFile(path, globID, serverMessages, r.server.MaxLineLength())
+			reader = &tailFile
+		}
+		limiter = r.server.TailLimiter()
 	}
 
+	// acquired tracks whether this goroutine successfully sent to the limiter.
+	// The defer must only release a slot when this goroutine actually holds one;
+	// an unconditional release would steal a slot from another goroutine that
+	// is still holding it, permanently reducing the effective semaphore capacity.
+	var acquired bool
 	defer func() {
-		select {
-		case <-limiter:
-		default:
+		if acquired {
+			<-limiter
 		}
 	}()
 
 	select {
 	case limiter <- struct{}{}:
+		acquired = true
+		dlog.Server.Debug(r.server.LogContext(), "Got limiter slot immediately", "path", path)
 	case <-ctx.Done():
+		dlog.Server.Debug(r.server.LogContext(), "Context cancelled while waiting for limiter", "path", path)
 		return
 	default:
-		dlog.Server.Info("Server limit hit, queueing file", len(limiter), path)
+		dlog.Server.Info(r.server.LogContext(), "Server limit hit, queueing file", "limiterLen", len(limiter), "path", path, "maxConcurrent", cap(limiter))
 		select {
 		case limiter <- struct{}{}:
-			dlog.Server.Info("Server limit OK now, processing file", len(limiter), path)
+			acquired = true
+			dlog.Server.Info(r.server.LogContext(), "Server limit OK now, processing file", "limiterLen", len(limiter), "path", path)
 		case <-ctx.Done():
+			dlog.Server.Debug(r.server.LogContext(), "Context cancelled while queued for limiter", "path", path)
 			return
 		}
 	}
 
-	lines := r.server.lines
-	aggregate := r.server.aggregate
+	// Output is the one and only read path. read() is only ever invoked for the
+	// cat/grep/tail command handlers (see makeReadCommandHandler), and MapReduce
+	// always builds a Aggregate for both server mode and serverless (see
+	// mapcommand.go). So either the read runs in cat/grep mode with
+	// Aggregate() non-nil and feeds it directly via AggregateProcessor,
+	// or it is genuine cat/grep/tail output using the output direct-output writer;
+	// makeProcessor picks between the two. The former channel-based fallback
+	// (readViaChannels feeding the regular server.Aggregate) was removed once
+	// serverless MapReduce migrated to the output aggregate (tasks sv0/hv0), so
+	// there is no non-output path left here.
+	dlog.Server.Debug(r.server.LogContext(), "Selecting read mode",
+		"mode", r.mode, "hasAggregate", r.server.Aggregate() != nil)
+	dlog.Server.Info(r.server.LogContext(), "Using turbo mode for reading", path, "mode", r.mode, "hasAggregate", r.server.Aggregate() != nil)
+	r.readWithProcessor(ctx, ltx, path, globID, re, reader)
+}
+
+func journalArgs(spec string) []string {
+	source := strings.TrimPrefix(spec, fs.JournalSpecPrefix)
+	if source == "" {
+		return nil
+	}
+	return []string{"-u", source}
+}
+
+func (r *readCommand) readWithProcessor(ctx context.Context, ltx lcontext.LContext,
+	path, globID string, re regex.Regex, reader fs.FileReader) {
+
+	dlog.Server.Info(r.server.LogContext(), "Using output channel-less implementation", path, globID)
+	r.logRegexMode(re)
+
+	r.ensureOutputEnabled(ctx)
+	writer := r.makeWriter(ctx)
+
+	r.executeReadLoop(ctx, ltx, path, globID, re, reader, r.readViaProcessor(path, globID, writer))
+}
+
+func (r *readCommand) executeReadLoop(ctx context.Context, ltx lcontext.LContext,
+	path, globID string, re regex.Regex, reader fs.FileReader, strategy readStrategy) {
 
 	for {
-		if aggregate != nil {
-			lines = make(chan *line.Line, 100)
-			aggregate.NextLinesCh <- lines
-		}
-		if err := reader.Start(ctx, ltx, lines, re); err != nil {
-			dlog.Server.Error(r.server.user, path, globID, err)
-		}
-		if aggregate != nil {
-			// Also makes aggregate to Flush
-			close(lines)
+		if err := strategy(ctx, ltx, reader, re); err != nil {
+			dlog.Server.Error(r.server.LogContext(), path, globID, err)
 		}
 
 		select {
@@ -184,12 +431,135 @@ func (r *readCommand) read(ctx context.Context, ltx lcontext.LContext,
 				return
 			}
 		}
-		time.Sleep(time.Second * 2)
+
+		if !ctxutil.Sleep(ctx, r.server.ReadRetryInterval()) {
+			return
+		}
 		dlog.Server.Info(path, globID, "Reading file again")
 	}
 }
 
-func (r *readCommand) makeGlobID(path, glob string) string {
+func (r *readCommand) readViaProcessor(path, globID string, writer LineWriter) readStrategy {
+	return func(ctx context.Context, ltx lcontext.LContext, reader fs.FileReader, re regex.Regex) error {
+		dlog.Server.Trace(r.server.LogContext(), path, globID, "readWithProcessor -> starting read loop iteration")
+
+		processor := r.makeProcessor(path, globID, writer)
+
+		dlog.Server.Trace(r.server.LogContext(), path, globID, "readWithProcessor -> reader.StartWithPocessorOptimized -> about to start")
+		startErr := reader.StartWithProcessorOptimized(ctx, ltx, processor, re)
+		dlog.Server.Trace(r.server.LogContext(), path, globID, "readWithProcessor -> reader.StartWithPocessorOptimized -> completed")
+
+		// Ensure we flush and close the processor before retry checks.
+		dlog.Server.Trace(r.server.LogContext(), path, globID, "readWithProcessor -> flushing processor")
+		if flushErr := processor.Flush(); flushErr != nil {
+			dlog.Server.Error(r.server.LogContext(), path, globID, "flush error", flushErr)
+		}
+		dlog.Server.Trace(r.server.LogContext(), path, globID, "readWithProcessor -> closing processor")
+		if closeErr := processor.Close(); closeErr != nil {
+			dlog.Server.Error(r.server.LogContext(), path, globID, "close error", closeErr)
+		}
+		dlog.Server.Trace(r.server.LogContext(), path, globID, "readWithProcessor -> processor closed")
+
+		// Give time for data to be transmitted.
+		// This is crucial for integration tests to ensure all data is sent
+		// Skip this delay in serverless mode since data is written directly to stdout
+		if !r.server.Serverless() {
+			dlog.Server.Trace(r.server.LogContext(), path, globID, "readWithProcessor -> waiting for data transmission")
+			if !ctxutil.Sleep(ctx, r.server.OutputTransmissionDelay()) {
+				return startErr
+			}
+		}
+
+		return startErr
+	}
+}
+
+func (r *readCommand) ensureOutputEnabled(ctx context.Context) {
+	// EnableDirectOutput is an atomic check-and-enable guarded by the output
+	// manager's mutex, so mode and channel initialization are always observed
+	// together — no visibility double-check is needed here. It returns false
+	// when output mode was already active.
+	if !r.server.EnableDirectOutput() {
+		return
+	}
+	// Wake a potentially blocked reader goroutine so it can switch to output drain path.
+	r.sendServerMessage(ctx, ".output wake")
+}
+
+func (r *readCommand) makeWriter(ctx context.Context) LineWriter {
+	// Create a writer instance per file to keep concurrent processing isolated.
+	if r.server.Serverless() {
+		return NewGeneratedDirectWriter(serverlessOutputWriter(), r.server.Hostname(), r.server.PlainOutput(), r.server.Serverless(), r.generation, r.server.ActiveSessionGeneration)
+	}
+
+	// Use NewNetworkWriter so bufSize is set to 64KB. A bare struct literal
+	// here previously left bufSize at zero, which disabled write batching and
+	// sent every line as its own output-channel payload (one SSH packet + one
+	// write syscall per line), making server-mode output output far slower than
+	// it should be.
+	return NewNetworkWriter(ctx, r.server.GetOutputChannel(),
+		r.server.ServerMessagesChannel(), r.server.Hostname(),
+		r.server.PlainOutput(), r.server.Serverless(), r.generation,
+		r.server.ActiveSessionGeneration)
+}
+
+// serverlessOutputWriter returns the io.Writer the serverless direct-output
+// (output) path writes payload to. It is os.Stdout by default. When the client
+// opted in via --log-payload / Client.LogPayload, it additionally tees the exact
+// same payload bytes into the fout daily log FILE sink.
+//
+// This is needed because output is now the only runtime path: the serverless
+// direct-output path writes payload straight to stdout and bypasses the fout
+// logger's Raw method, so the logger's own --log-payload file tee never runs.
+// The tee is added via io.MultiWriter, which writes to os.Stdout first and byte
+// for byte unchanged, so stdout stays identical whether or not --log-payload is
+// set; only the file gains the payload. When LogPayload is off (the default) we
+// return the bare os.Stdout so the hot path stays allocation-free.
+func serverlessOutputWriter() io.Writer {
+	if config.Client != nil && config.Client.LogPayload {
+		return io.MultiWriter(os.Stdout, payloadFileTeeWriter{})
+	}
+	return os.Stdout
+}
+
+// payloadFileTeeWriter is an io.Writer that forwards serverless payload bytes to
+// the client logger's FILE sink only (never stdout), honoring --log-payload. It
+// lets the serverless direct-output path reuse the fout daily-log file tee that
+// the bypassed logger.Raw path would otherwise have provided.
+type payloadFileTeeWriter struct{}
+
+func (payloadFileTeeWriter) Write(p []byte) (int, error) {
+	// dlog.Client is the fout logger that owns the daily log file in serverless
+	// mode; RawPayloadFileTee no-ops when the logger has no file sink or payload
+	// teeing is disabled. Report the full length as written so io.MultiWriter
+	// does not treat the tee as a short write.
+	if dlog.Client != nil {
+		dlog.Client.RawPayloadFileTee(string(p))
+	}
+	return len(p), nil
+}
+
+func (r *readCommand) makeProcessor(path, globID string, writer LineWriter) readProcessor {
+	if aggregate := r.server.Aggregate(); aggregate != nil {
+		dlog.Server.Info(r.server.LogContext(), "Using turbo aggregate processor for MapReduce", path, globID)
+		return server.NewAggregateProcessor(aggregate, globID)
+	}
+
+	return NewDirectLineProcessor(writer, globID)
+}
+
+func (r *readCommand) logRegexMode(re regex.Regex) {
+	if r.mode != omode.GrepClient {
+		return
+	}
+	if re.IsLiteral() {
+		dlog.Server.Info(r.server.LogContext(), "Using optimized literal string matching for pattern:", re.Pattern())
+	} else {
+		dlog.Server.Info(r.server.LogContext(), "Using regex matching for pattern:", re.Pattern())
+	}
+}
+
+func (r *readCommand) makeGlobID(ctx context.Context, path, glob string) string {
 	var idParts []string
 	pathParts := strings.Split(path, "/")
 
@@ -206,13 +576,54 @@ func (r *readCommand) makeGlobID(path, glob string) string {
 		return pathParts[len(pathParts)-1]
 	}
 
-	r.server.sendln(r.server.serverMessages,
-		dlog.Server.Warn("Empty file path given?", path, glob))
+	r.sendServerMessage(ctx, dlog.Server.Warn("Empty file path given?", path, glob))
 	return ""
 }
 
+// sendServerMessage forwards a user-visible message to the session's shared
+// serverMessages channel (capacity 10, drained only by baseHandler.Read). A
+// bare send here would block forever once the client disconnects and Read
+// stops draining — pinning one goroutine per stuck send (e.g. one per
+// permission-denied file of a glob expansion, up to MaxGlobTargets). The
+// select on ctx.Done() makes the send abandonable: the per-command context is
+// cancelled on command completion and on handler shutdown (see
+// baseHandler.newCommandContext), mirroring the done-guarded baseHandler.send.
+func (r *readCommand) sendServerMessage(ctx context.Context, message string) {
+	select {
+	case r.server.ServerMessagesChannel() <- encodeGeneratedMessage(r.generation, message+"\n"):
+	case <-ctx.Done():
+	}
+}
+
+func (r *readCommand) newGeneratedServerMessagesChannel(ctx context.Context) (chan string, func()) {
+	serverMessages := make(chan string, 16)
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		for {
+			select {
+			case message, ok := <-serverMessages:
+				if !ok {
+					return
+				}
+				select {
+				case r.server.ServerMessagesChannel() <- encodeGeneratedMessage(r.generation, message):
+				case <-ctx.Done():
+					return
+				}
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+	return serverMessages, func() {
+		close(serverMessages)
+		<-done
+	}
+}
+
 func (r *readCommand) isInputFromPipe() bool {
-	if !r.server.serverless {
+	if !r.server.Serverless() {
 		// Can read from pipe only in serverless mode.
 		return false
 	}