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

diff --git a/internal/config/server.go b/internal/config/server.go
index cb9ca2b..b967103 100644
--- a/internal/config/server.go
+++ b/internal/config/server.go
@@ -67,6 +67,50 @@ type ServerConfig struct {
 	Ciphers []string `json:",omitempty"`
 	// The allowed MAC algorithms.
 	MACs []string `json:",omitempty"`
+	// Enable in-memory auth-key registration and fast reconnect.
+	AuthKeyEnabled bool `json:",omitempty"`
+	// Auth-key cache entry TTL in seconds.
+	AuthKeyTTLSeconds int `json:",omitempty"`
+	// Maximum number of cached auth keys per user.
+	AuthKeyMaxPerUser int `json:",omitempty"`
+	// Retry interval for glob retries in milliseconds.
+	ReadGlobRetryIntervalMs int `json:",omitempty"`
+	// Retry interval for re-reading in tail/cat loops in milliseconds.
+	ReadRetryIntervalMs int `json:",omitempty"`
+	// Delay after output processor flush/close to allow data transmission, in milliseconds.
+	OutputTransmissionDelayMs int `json:",omitempty"`
+	// Output EOF wait base duration in milliseconds.
+	OutputEOFWaitBaseMs int `json:",omitempty"`
+	// Output EOF wait per-file duration in milliseconds.
+	OutputEOFWaitPerFileMs int `json:",omitempty"`
+	// Maximum output EOF wait duration in milliseconds.
+	OutputEOFWaitMaxMs int `json:",omitempty"`
+	// Output channel buffer size.
+	OutputChannelBufferSize int `json:",omitempty"`
+	// Output channel flush timeout in milliseconds.
+	OutputFlushTimeoutMs int `json:",omitempty"`
+	// Output channel flush poll interval in milliseconds.
+	OutputFlushPollIntervalMs int `json:",omitempty"`
+	// Output read retry interval in milliseconds when data is expected but not yet available.
+	OutputReadRetryIntervalMs int `json:",omitempty"`
+	// Maximum time to wait for output EOF acknowledgement after signaling EOF, in milliseconds.
+	OutputEOFAckTimeoutMs int `json:",omitempty"`
+	// Wait for aggregate serialization during shutdown in milliseconds.
+	ShutdownOutputSerializeWaitMs int `json:",omitempty"`
+	// Final idle recheck wait before shutdown in milliseconds.
+	ShutdownIdleRecheckWaitMs int `json:",omitempty"`
+	// Maximum size in bytes of a single command frame (bytes accumulated between
+	// ';' delimiters). Frames that grow beyond this limit are rejected and the
+	// session is closed to prevent unbounded memory exhaustion by a malicious or
+	// misbehaving client. Default is 1 MiB.
+	MaxCommandFrameSize int `json:",omitempty"`
+	// Maximum number of glob expansion targets (file paths) that a single read
+	// command is allowed to dispatch. When a glob pattern expands to more paths
+	// than this limit, the excess paths are dropped and a warning is sent to the
+	// client. This prevents an authenticated user with broad read permission from
+	// spawning unbounded goroutines and exhausting server memory/CPU.
+	// Default is 1000. Set to 0 to keep the built-in default.
+	MaxGlobTargets int `json:",omitempty"`
 }
 
 // Create a new default server configuration.
@@ -85,13 +129,42 @@ func newDefaultServerConfig() *ServerConfig {
 		Permissions: Permissions{
 			Default: defaultPermissions,
 		},
+		AuthKeyEnabled:                true,
+		AuthKeyTTLSeconds:             86400,
+		AuthKeyMaxPerUser:             5,
+		ReadGlobRetryIntervalMs:       5000,
+		ReadRetryIntervalMs:           2000,
+		OutputTransmissionDelayMs:     50,
+		OutputEOFWaitBaseMs:           500,
+		OutputEOFWaitPerFileMs:        10,
+		OutputEOFWaitMaxMs:            2000,
+		OutputChannelBufferSize:       1000,
+		OutputFlushTimeoutMs:          2000,
+		OutputFlushPollIntervalMs:     10,
+		OutputReadRetryIntervalMs:     1,
+		OutputEOFAckTimeoutMs:         2000,
+		ShutdownOutputSerializeWaitMs: 500,
+		ShutdownIdleRecheckWaitMs:     10,
+		MaxCommandFrameSize:           DefaultMaxCommandFrameSize,
+		MaxGlobTargets:                1000,
 	}
 }
 
-// ServerUserPermissions retrieves the permission set of a given user.
-func ServerUserPermissions(userName string) (permissions []string, err error) {
-	permissions = Server.Permissions.Default
-	if p, ok := Server.Permissions.Users[userName]; ok {
+// NewDefaultServerConfigForTest returns a fresh ServerConfig populated with all
+// default values. It is intended for use in unit tests that need to inspect or
+// compare default configuration without running the full config initializer.
+func NewDefaultServerConfigForTest() *ServerConfig {
+	return newDefaultServerConfig()
+}
+
+// UserPermissions retrieves the permission set of a given user.
+func (c *ServerConfig) UserPermissions(userName string) (permissions []string, err error) {
+	if c == nil {
+		return nil, errors.New("missing server config")
+	}
+
+	permissions = c.Permissions.Default
+	if p, ok := c.Permissions.Users[userName]; ok {
 		permissions = p
 	}
 	if len(permissions) == 0 {
@@ -99,3 +172,11 @@ func ServerUserPermissions(userName string) (permissions []string, err error) {
 	}
 	return
 }
+
+// ServerUserPermissions retrieves the permission set of a given user.
+func ServerUserPermissions(userName string) (permissions []string, err error) {
+	if Server == nil {
+		return nil, errors.New("missing server config")
+	}
+	return Server.UserPermissions(userName)
+}