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, 167 insertions, 3 deletions

diff --git a/internal/mapr/logformat/csv_test.go b/internal/mapr/logformat/csv_test.go
index 1baf032..fa85a99 100644
--- a/internal/mapr/logformat/csv_test.go
+++ b/internal/mapr/logformat/csv_test.go
@@ -2,6 +2,7 @@ package logformat
 
 import (
 	"strings"
+	"sync"
 	"testing"
 
 	"github.com/mimecast/dtail/internal/protocol"
@@ -23,13 +24,15 @@ func TestCSVLogFormat(t *testing.T) {
 		strings.Join(dataLine2, protocol.CSVDelimiter),
 	}
 
+	const sourceID = "file-a"
+
 	// First line is the header!
-	if _, err := parser.MakeFields(inputs[0]); err != ErrIgnoreFields {
+	if _, err := parser.MakeFields(inputs[0], sourceID); err != ErrIgnoreFields {
 		t.Errorf("Unable to parse the CSV header")
 	}
 
 	// First data line
-	fields, err := parser.MakeFields(inputs[1])
+	fields, err := parser.MakeFields(inputs[1], sourceID)
 	if err != nil {
 		t.Errorf("Unable to parse first CSV data line: %s", err.Error())
 	}
@@ -41,7 +44,7 @@ func TestCSVLogFormat(t *testing.T) {
 	}
 
 	// Second data line
-	fields, err = parser.MakeFields(inputs[2])
+	fields, err = parser.MakeFields(inputs[2], sourceID)
 	if err != nil {
 		t.Errorf("Unable to parse first CSV data line: %s", err.Error())
 	}
@@ -52,3 +55,164 @@ func TestCSVLogFormat(t *testing.T) {
 		t.Errorf("Expected 'color' to be 'Black' but got '%s'", val)
 	}
 }
+
+// TestCSVLogFormatMultiFileHeaders reproduces the bug where a single
+// csvParser instance (as used by the Aggregate for every file
+// in a mapreduce session) was treating the header row of every file after
+// the first as a data row, silently corrupting aggregates.
+func TestCSVLogFormatMultiFileHeaders(t *testing.T) {
+	parser, err := NewParser("csv", nil)
+	if err != nil {
+		t.Fatalf("Unable to create parser: %s", err.Error())
+	}
+
+	headersA := []string{"name", "value"}
+	headersB := []string{"color", "count"}
+
+	fileA := []string{
+		strings.Join(headersA, protocol.CSVDelimiter),
+		strings.Join([]string{"alpha", "1"}, protocol.CSVDelimiter),
+		strings.Join([]string{"beta", "2"}, protocol.CSVDelimiter),
+	}
+	fileB := []string{
+		strings.Join(headersB, protocol.CSVDelimiter),
+		strings.Join([]string{"orange", "3"}, protocol.CSVDelimiter),
+		strings.Join([]string{"black", "4"}, protocol.CSVDelimiter),
+	}
+
+	const sourceA = "file-a"
+	const sourceB = "file-b"
+
+	// First line of file A is its header.
+	if _, err := parser.MakeFields(fileA[0], sourceA); err != ErrIgnoreFields {
+		t.Fatalf("Expected header line of file A to be ignored, got err=%v", err)
+	}
+	for _, line := range fileA[1:] {
+		fields, err := parser.MakeFields(line, sourceA)
+		if err != nil {
+			t.Fatalf("Unable to parse data line %q of file A: %s", line, err.Error())
+		}
+		if _, ok := fields["name"]; !ok {
+			t.Errorf("Expected file A field 'name' for line %q, got %v", line, fields)
+		}
+	}
+
+	// First line of file B MUST also be treated as a header, not a data row.
+	if _, err := parser.MakeFields(fileB[0], sourceB); err != ErrIgnoreFields {
+		t.Fatalf("Expected header line of file B to be ignored (bug: header is being consumed as a data row), got err=%v", err)
+	}
+
+	// Data lines of file B must be mapped against file B's headers, not
+	// file A's.
+	for _, line := range fileB[1:] {
+		fields, err := parser.MakeFields(line, sourceB)
+		if err != nil {
+			t.Fatalf("Unable to parse data line %q of file B: %s", line, err.Error())
+		}
+		if _, ok := fields["color"]; !ok {
+			t.Errorf("Expected file B field 'color' for line %q, got %v", line, fields)
+		}
+		if _, ok := fields["name"]; ok {
+			t.Errorf("File B line %q should not carry file A field 'name'; got %v",
+				line, fields)
+		}
+	}
+}
+
+// TestCSVLogFormatConcurrentSameSourceInstall reproduces a TOCTOU bug in
+// csvParser.MakeFields: the original code first called headerFor under
+// RLock, and only if the header was missing did it call parseHeader under
+// Lock. Two goroutines racing on the same sourceID could both observe
+// "missing" and both return ErrIgnoreFields, silently dropping the loser's
+// data row (the installer wrote the header; the non-installer still
+// signalled "this line was a header" to the caller).
+//
+// With the fix, the check-and-install is a single critical section and
+// exactly one of the two concurrent calls reports ErrIgnoreFields; the
+// other maps its line against the installed header.
+func TestCSVLogFormatConcurrentSameSourceInstall(t *testing.T) {
+	header := strings.Join([]string{"name", "value"}, protocol.CSVDelimiter)
+	data := strings.Join([]string{"alpha", "1"}, protocol.CSVDelimiter)
+
+	const attempts = 500
+	const sourceID = "file-race"
+
+	for attempt := 0; attempt < attempts; attempt++ {
+		parser, err := NewParser("csv", nil)
+		if err != nil {
+			t.Fatalf("attempt %d: unable to create parser: %s", attempt, err.Error())
+		}
+
+		lines := [2]string{header, data}
+		var results [2]struct {
+			fields map[string]string
+			err    error
+		}
+
+		start := make(chan struct{})
+		var wg sync.WaitGroup
+		wg.Add(2)
+		for i := 0; i < 2; i++ {
+			i := i
+			go func() {
+				defer wg.Done()
+				<-start
+				results[i].fields, results[i].err = parser.MakeFields(lines[i], sourceID)
+			}()
+		}
+		close(start)
+		wg.Wait()
+
+		ignored := 0
+		for _, r := range results {
+			if r.err == ErrIgnoreFields {
+				ignored++
+			}
+		}
+		if ignored != 1 {
+			t.Fatalf("attempt %d: expected exactly one ErrIgnoreFields across two racing calls on the same sourceID, got %d; results=%+v",
+				attempt, ignored, results)
+		}
+	}
+}
+
+// TestCSVLogFormatConcurrentSources ensures the per-source header store is
+// safe for concurrent access across multiple sourceIDs, matching how the
+// aggregator drives the parser from batched lines across files.
+func TestCSVLogFormatConcurrentSources(t *testing.T) {
+	parser, err := NewParser("csv", nil)
+	if err != nil {
+		t.Fatalf("Unable to create parser: %s", err.Error())
+	}
+
+	header := strings.Join([]string{"name", "value"}, protocol.CSVDelimiter)
+	data := strings.Join([]string{"alpha", "1"}, protocol.CSVDelimiter)
+
+	const workers = 16
+	const iterations = 64
+
+	var wg sync.WaitGroup
+	wg.Add(workers)
+	for w := 0; w < workers; w++ {
+		go func(id int) {
+			defer wg.Done()
+			sourceID := "source-" + string(rune('a'+id))
+			if _, err := parser.MakeFields(header, sourceID); err != ErrIgnoreFields {
+				t.Errorf("worker %d: expected header to be ignored, got err=%v", id, err)
+				return
+			}
+			for i := 0; i < iterations; i++ {
+				fields, err := parser.MakeFields(data, sourceID)
+				if err != nil {
+					t.Errorf("worker %d: parse err=%v", id, err)
+					return
+				}
+				if fields["name"] != "alpha" {
+					t.Errorf("worker %d: expected name=alpha, got %q", id, fields["name"])
+					return
+				}
+			}
+		}(w)
+	}
+	wg.Wait()
+}