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, 43 insertions, 4 deletions

diff --git a/internal/config/args.go b/internal/config/args.go
index 87ef393..f0b14e1 100644
--- a/internal/config/args.go
+++ b/internal/config/args.go
@@ -3,6 +3,7 @@ package config
 import (
 	"encoding/base64"
 	"fmt"
+	"sort"
 	"strconv"
 	"strings"
 
@@ -18,16 +19,21 @@ type Args struct {
 	Arguments             []string
 	ConfigFile            string
 	ConnectionsPerCPU     int
+	ControlTTYPath        string
 	Discovery             string
+	InteractiveQuery      bool
 	LogDir                string
 	Logger                string
 	LogLevel              string
+	LogPayload            bool
 	Mode                  omode.Mode
+	NoAuthKey             bool
 	NoColor               bool
 	QueryStr              string
 	Quiet                 bool
 	RegexInvert           bool
 	RegexStr              string
+	SSHAgentKeyIndex      int
 	SSHAuthMethods        []gossh.AuthMethod
 	SSHBindAddress        string
 	SSHHostKeyCallback    gossh.HostKeyCallback
@@ -50,16 +56,21 @@ func (a *Args) String() string {
 	sb.WriteString(fmt.Sprintf("%s:%v,", "Arguments", a.Arguments))
 	sb.WriteString(fmt.Sprintf("%s:%v,", "ConfigFile", a.ConfigFile))
 	sb.WriteString(fmt.Sprintf("%s:%v,", "ConnectionsPerCPU", a.ConnectionsPerCPU))
+	sb.WriteString(fmt.Sprintf("%s:%v,", "ControlTTYPath", a.ControlTTYPath))
 	sb.WriteString(fmt.Sprintf("%s:%v,", "Discovery", a.Discovery))
+	sb.WriteString(fmt.Sprintf("%s:%v,", "InteractiveQuery", a.InteractiveQuery))
 	sb.WriteString(fmt.Sprintf("%s:%v,", "LogDir", a.LogDir))
 	sb.WriteString(fmt.Sprintf("%s:%v,", "LogLevel", a.LogLevel))
+	sb.WriteString(fmt.Sprintf("%s:%v,", "LogPayload", a.LogPayload))
 	sb.WriteString(fmt.Sprintf("%s:%v,", "Logger", a.Logger))
 	sb.WriteString(fmt.Sprintf("%s:%v,", "Mode", a.Mode))
+	sb.WriteString(fmt.Sprintf("%s:%v,", "NoAuthKey", a.NoAuthKey))
 	sb.WriteString(fmt.Sprintf("%s:%v,", "NoColor", a.NoColor))
 	sb.WriteString(fmt.Sprintf("%s:%v,", "QueryStr", a.QueryStr))
 	sb.WriteString(fmt.Sprintf("%s:%v,", "Quiet", a.Quiet))
 	sb.WriteString(fmt.Sprintf("%s:%v,", "RegexInvert", a.RegexInvert))
 	sb.WriteString(fmt.Sprintf("%s:%v,", "RegexStr", a.RegexStr))
+	sb.WriteString(fmt.Sprintf("%s:%v,", "SSHAgentKeyIndex", a.SSHAgentKeyIndex))
 	sb.WriteString(fmt.Sprintf("%s:%v,", "SSHAuthMethods", a.SSHAuthMethods))
 	sb.WriteString(fmt.Sprintf("%s:%v,", "SSHBindAddress", a.SSHBindAddress))
 	sb.WriteString(fmt.Sprintf("%s:%v,", "SSHHostKeyCallback", a.SSHHostKeyCallback))
@@ -100,25 +111,53 @@ func (a *Args) SerializeOptions() string {
 		options["after"] = fmt.Sprintf("%d", a.LContext.AfterContext)
 	}
 
+	return serializeOptions(options)
+}
+
+func serializeOptions(options map[string]string) string {
+	if len(options) == 0 {
+		return ""
+	}
+
+	keys := make([]string, 0, len(options))
+	for k := range options {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+
 	var sb strings.Builder
-	var i int
-	for k, v := range options {
+	for i, k := range keys {
 		if i > 0 {
 			sb.WriteString(":")
 		}
 		sb.WriteString(k)
 		sb.WriteString("=")
-		sb.WriteString(v)
-		i++
+		sb.WriteString(serializeOptionValue(options[k]))
 	}
 	return sb.String()
 }
 
+func serializeOptionValue(value string) string {
+	if strings.ContainsAny(value, ":=|") || strings.HasPrefix(value, "base64%") {
+		return "base64%" + base64.StdEncoding.EncodeToString([]byte(value))
+	}
+
+	return value
+}
+
 // DeserializeOptions deserializes the options, but into a map.
 func DeserializeOptions(opts []string) (map[string]string, lcontext.LContext, error) {
 	options := make(map[string]string, len(opts))
 	var ltx lcontext.LContext
 
+	if len(opts) == 1 {
+		raw := strings.TrimSpace(opts[0])
+		if raw == "" {
+			return options, ltx, nil
+		}
+		opts = strings.Split(raw, ":")
+	}
+
 	for _, o := range opts {
 		kv := strings.SplitN(o, "=", 2)
 		if len(kv) != 2 {