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, 19 insertions, 10 deletions

diff --git a/internal/mapr/query.go b/internal/mapr/query.go
index 139f04c..5c22dec 100644
--- a/internal/mapr/query.go
+++ b/internal/mapr/query.go
@@ -19,7 +19,8 @@ type Outfile struct {
 	AppendMode bool
 }
 
-func (o Outfile) String() string {
+// String returns the string representation of Outfile.
+func (o *Outfile) String() string {
 	return fmt.Sprintf("Outfile(FilePath:%v,AppendMode:%v)", o.FilePath, o.AppendMode)
 }
 
@@ -41,7 +42,8 @@ type Query struct {
 	LogFormat    string
 }
 
-func (q Query) String() string {
+// String returns the string representation of Query.
+func (q *Query) String() string {
 	return fmt.Sprintf("Query(Select:%v,Table:%s,Where:%v,Set:%vGroupBy:%v,"+
 		"GroupKey:%s,OrderBy:%v,ReverseOrder:%v,Interval:%v,Limit:%d,Outfile:%s,"+
 		"RawQuery:%s,tokens:%v,LogFormat:%s)",
@@ -74,13 +76,20 @@ func NewQuery(queryStr string) (*Query, error) {
 		Limit:    -1,
 	}
 
-	// If log format is CSV, then use "." as the table. It means, that
-	// we don't do any file filtering, we process all lines of the CSV.
-	if q.LogFormat == "csv" {
+	// Parse the query tokens to populate all fields including LogFormat and Table.
+	if err := q.parse(tokens); err != nil {
+		return nil, err
+	}
+
+	// If the log format is CSV and no explicit FROM table was provided, default
+	// the table to "." so that all lines are processed without file filtering.
+	// This check must run after parse() because LogFormat is only populated
+	// once parseTokens() has processed the "logformat" keyword.
+	if q.LogFormat == "csv" && q.Table == "" {
 		q.Table = "."
 	}
 
-	return &q, q.parse(tokens)
+	return &q, nil
 }
 
 // HasOutfile returns true if query result will be written to a CVS output file.
@@ -95,7 +104,7 @@ func (q *Query) Has(what string) bool {
 
 func (q *Query) parse(tokens []token) error {
 	if _, err := q.parseTokens(tokens); err != nil {
-		return err
+		return fmt.Errorf("failed to parse query tokens: %w", err)
 	}
 
 	if len(q.Select) < 1 {
@@ -162,14 +171,14 @@ func (q *Query) parseTokens(tokens []token) ([]token, error) {
 			}
 		case "group":
 			tokens = tokensConsumeOptional(tokens[1:], "by")
-			if tokens == nil || len(tokens) < 1 {
+			if len(tokens) < 1 {
 				return tokens, errors.New(invalidQuery + unexpectedEnd)
 			}
 			tokens, q.GroupBy = tokensConsumeStr(tokens)
 			q.GroupKey = strings.Join(q.GroupBy, ",")
 		case "rorder":
 			tokens = tokensConsumeOptional(tokens[1:], "by")
-			if tokens == nil || len(tokens) < 1 {
+			if len(tokens) < 1 {
 				return tokens, errors.New(invalidQuery + unexpectedEnd)
 			}
 			tokens, found = tokensConsume(tokens)
@@ -180,7 +189,7 @@ func (q *Query) parseTokens(tokens []token) ([]token, error) {
 			q.ReverseOrder = true
 		case "order":
 			tokens = tokensConsumeOptional(tokens[1:], "by")
-			if tokens == nil || len(tokens) < 1 {
+			if len(tokens) < 1 {
 				return tokens, errors.New(invalidQuery + unexpectedEnd)
 			}
 			tokens, found = tokensConsume(tokens)