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, 148 insertions, 78 deletions

diff --git a/internal/io/dlog/dlog.go b/internal/io/dlog/dlog.go
index 258fb68..e28b442 100644
--- a/internal/io/dlog/dlog.go
+++ b/internal/io/dlog/dlog.go
@@ -31,36 +31,6 @@ var Common *DLog
 var mutex sync.Mutex
 var started bool
 
-// Start logger(s).
-func Start(ctx context.Context, wg *sync.WaitGroup, sourceProcess source.Source) {
-	mutex.Lock()
-	defer mutex.Unlock()
-
-	if started {
-		Common.FatalPanic("Logger already started")
-	}
-
-	Client = new(sourceProcess, source.Client)
-	Server = new(sourceProcess, source.Server)
-	Common = Client
-	if sourceProcess == source.Server {
-		Common = Server
-	}
-
-	var wg2 sync.WaitGroup
-	wg2.Add(2)
-	go Client.start(ctx, &wg2)
-	go Server.start(ctx, &wg2)
-
-	go rotation(ctx)
-	go func() {
-		wg2.Wait()
-		wg.Done()
-	}()
-
-	started = true
-}
-
 // DLog is the DTail logger.
 type DLog struct {
 	logger loggers.Logger
@@ -94,62 +64,43 @@ func new(sourceProcess, sourcePackage source.Source) *DLog {
 	}
 }
 
-func (d *DLog) start(ctx context.Context, wg *sync.WaitGroup) {
-	defer wg.Done()
-	var wg2 sync.WaitGroup
-	wg2.Add(1)
-	d.logger.Start(ctx, &wg2)
-	<-ctx.Done()
-	wg2.Wait()
-}
+// Start logger(s).
+func Start(ctx context.Context, wg *sync.WaitGroup, sourceProcess source.Source) {
+	mutex.Lock()
+	defer mutex.Unlock()
 
-func (d *DLog) log(level level, args []interface{}) string {
-	if d.maxLevel < level {
-		return ""
+	if started {
+		Common.FatalPanic("Logger already started")
 	}
-	sb := pool.BuilderBuffer.Get().(*strings.Builder)
-	defer pool.RecycleBuilderBuffer(sb)
-	now := time.Now()
 
-	switch d.sourceProcess {
-	case source.Client:
-		sb.WriteString(d.sourcePackage.String())
-		sb.WriteString(protocol.FieldDelimiter)
-		sb.WriteString(d.hostname)
-		sb.WriteString(protocol.FieldDelimiter)
-		sb.WriteString(level.String())
-	default:
-		sb.WriteString(level.String())
-		sb.WriteString(protocol.FieldDelimiter)
-		sb.WriteString(now.Format("0102-150405"))
+	Client = new(sourceProcess, source.Client)
+	Server = new(sourceProcess, source.Server)
+	Common = Client
+	if sourceProcess == source.Server {
+		Common = Server
 	}
-	sb.WriteString(protocol.FieldDelimiter)
-	d.writeArgStrings(sb, args)
 
-	message := sb.String()
-	if !config.Client.TermColorsEnable || !d.logger.SupportsColors() {
-		d.logger.Log(now, message)
-		return message
-	}
+	var wg2 sync.WaitGroup
+	wg2.Add(2)
+	go Client.start(ctx, &wg2)
+	go Server.start(ctx, &wg2)
 
-	d.logger.LogWithColors(now, message, brush.Colorfy(message))
-	return message
+	go rotation(ctx)
+	go func() {
+		wg2.Wait()
+		wg.Done()
+	}()
+
+	started = true
 }
 
-func (d *DLog) writeArgStrings(sb *strings.Builder, args []interface{}) {
-	for i, arg := range args {
-		if i > 0 {
-			sb.WriteString(protocol.FieldDelimiter)
-		}
-		switch v := arg.(type) {
-		case string:
-			sb.WriteString(v)
-		case error:
-			sb.WriteString(v.Error())
-		default:
-			sb.WriteString(fmt.Sprintf("%v", v))
-		}
-	}
+func (d *DLog) start(ctx context.Context, wg *sync.WaitGroup) {
+	defer wg.Done()
+	var wg2 sync.WaitGroup
+	wg2.Add(1)
+	d.logger.Start(ctx, &wg2)
+	<-ctx.Done()
+	wg2.Wait()
 }
 
 // FatalPanic terminates the process with a fatal error.
@@ -192,8 +143,34 @@ func (d *DLog) Debug(args ...interface{}) string {
 	return d.log(Debug, args)
 }
 
+// TraceEnabled reports whether trace-level logging is currently active.
+//
+// It performs exactly the same maxLevel comparison as Trace's internal
+// early-return (see below), letting callers on per-line hot paths gate the
+// whole trace call — the variadic []interface{} slice allocation plus the
+// interface boxing of every non-pointer argument (uint64 line counts via
+// runtime.convT64, strings via convTstring) — behind one cheap, inlinable,
+// allocation-free branch. Without this guard those args are boxed at the call
+// site before Trace even runs, so Trace's own early-return cannot save them.
+//
+// The receiver is nil-safe so call sites need no separate nil check on the
+// package-level loggers (Server/Client/Common), which stay nil until Start.
+// maxLevel is fixed at logger construction from config.Common.LogLevel, so the
+// result mirrors whatever level Trace itself would observe.
+func (d *DLog) TraceEnabled() bool {
+	return d != nil && d.maxLevel >= Trace
+}
+
 // Trace logging.
 func (d *DLog) Trace(args ...interface{}) string {
+	// Early check to avoid expensive runtime.Caller when trace is disabled
+	// This is a critical performance optimization for hot paths. Note that on
+	// per-line hot paths callers should additionally gate with TraceEnabled()
+	// so the argument boxing never happens; this check only saves runtime.Caller
+	// and the log formatting, not the caller-side boxing of args.
+	if d.maxLevel < Trace {
+		return ""
+	}
 	_, file, line, _ := runtime.Caller(1)
 	args = append(args, fmt.Sprintf("at %s:%d", file, line))
 	return d.log(Trace, args)
@@ -201,6 +178,10 @@ func (d *DLog) Trace(args ...interface{}) string {
 
 // Devel used for development purpose only logging (e.g. "print" debugging).
 func (d *DLog) Devel(args ...interface{}) string {
+	// Early check to avoid expensive runtime.Caller when devel is disabled
+	if d.maxLevel < Devel {
+		return ""
+	}
 	_, file, line, _ := runtime.Caller(1)
 	args = append(args, fmt.Sprintf("at %s:%d", file, line))
 	return d.log(Devel, args)
@@ -216,6 +197,46 @@ func (d *DLog) Raw(message string) string {
 	return message
 }
 
+// payloadFileTeer is the optional capability of a logger that can tee retrieved
+// payload into its FILE sink without also writing it to stdout. Only the default
+// fout logger implements it; stdout/none loggers have no file sink and are
+// skipped via the type assertion in RawPayloadFileTee.
+type payloadFileTeer interface {
+	RawFileOnly(now time.Time, message string)
+}
+
+// RawPayloadFileTee writes retrieved payload to the logger's FILE sink only
+// (never stdout), honoring the client's --log-payload / Client.LogPayload
+// opt-in. It is used by the serverless direct-output path, which emits payload
+// straight to stdout and therefore bypasses the fout logger's own Raw tee. When
+// the active logger has no file sink (stdout/none) or payload teeing is
+// disabled, this is a no-op. stdout output is unaffected: the caller writes the
+// same payload bytes to stdout itself, and this method only adds the file tee.
+func (d *DLog) RawPayloadFileTee(message string) {
+	if teer, ok := d.logger.(payloadFileTeer); ok {
+		teer.RawFileOnly(time.Now(), message)
+	}
+}
+
+// RawLog writes a pre-formatted message through the DIAGNOSTIC (Log) sink, so it
+// reaches both stdout and — in the default fout logger — the daily log file.
+//
+// It differs from Raw, which uses the PAYLOAD sink: with the default
+// Client.LogPayload=false, Raw is gated out of the file (bulk dcat/dgrep/dtail
+// output must not grow the log). RawLog is for audit-worthy, already-formatted
+// lines such as server-error reports, which must always be kept in the file like
+// other diagnostics rather than being treated as bulk payload. The message is
+// written verbatim (no level/hostname prefix); callers pre-format it and must
+// NOT append a trailing newline, since the Log sink appends one.
+func (d *DLog) RawLog(message string) string {
+	if !config.Client.TermColorsEnable || !d.logger.SupportsColors() {
+		d.logger.Log(time.Now(), message)
+		return message
+	}
+	d.logger.LogWithColors(time.Now(), message, brush.Colorfy(message))
+	return message
+}
+
 // Mapreduce logging.
 func (d *DLog) Mapreduce(table string, data map[string]interface{}) string {
 	args := make([]interface{}, len(data)+1)
@@ -269,3 +290,52 @@ func (d *DLog) Pause() { d.logger.Pause() }
 
 // Resume the logging.
 func (d *DLog) Resume() { d.logger.Resume() }
+
+func (d *DLog) log(level level, args []interface{}) string {
+	if d.maxLevel < level {
+		return ""
+	}
+	sb := pool.BuilderBuffer.Get().(*strings.Builder)
+	defer pool.RecycleBuilderBuffer(sb)
+	now := time.Now()
+
+	switch d.sourceProcess {
+	case source.Client:
+		sb.WriteString(d.sourcePackage.String())
+		sb.WriteString(protocol.FieldDelimiter)
+		sb.WriteString(d.hostname)
+		sb.WriteString(protocol.FieldDelimiter)
+		sb.WriteString(level.String())
+	default:
+		sb.WriteString(level.String())
+		sb.WriteString(protocol.FieldDelimiter)
+		sb.WriteString(now.Format("0102-150405"))
+	}
+	sb.WriteString(protocol.FieldDelimiter)
+	d.writeArgStrings(sb, args)
+
+	message := sb.String()
+	if !config.Client.TermColorsEnable || !d.logger.SupportsColors() {
+		d.logger.Log(now, message)
+		return message
+	}
+
+	d.logger.LogWithColors(now, message, brush.Colorfy(message))
+	return message
+}
+
+func (d *DLog) writeArgStrings(sb *strings.Builder, args []interface{}) {
+	for i, arg := range args {
+		if i > 0 {
+			sb.WriteString(protocol.FieldDelimiter)
+		}
+		switch v := arg.(type) {
+		case string:
+			sb.WriteString(v)
+		case error:
+			sb.WriteString(v.Error())
+		default:
+			sb.WriteString(fmt.Sprintf("%v", v))
+		}
+	}
+}