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, 77 insertions, 37 deletions

diff --git a/internal/mapr/funcs/function_test.go b/internal/mapr/funcs/function_test.go
index 8b5d8b7..8227817 100644
--- a/internal/mapr/funcs/function_test.go
+++ b/internal/mapr/funcs/function_test.go
@@ -2,51 +2,91 @@ package funcs
 
 import "testing"
 
-func TestFunction(t *testing.T) {
-	input := "md5sum($line)"
-	fs, arg, err := NewFunctionStack(input)
-	if err != nil {
-		t.Errorf("error parsing function input '%s': %s (%v)\n",
-			input, err.Error(), fs)
-	}
-	if arg != "$line" {
-		t.Errorf("error parsing function input '%s': expected argument '$line' but "+
-			"got '%s' (%v)\n", input, arg, fs)
-	}
-	t.Log(input, fs, arg)
+func TestFunctionStackValid(t *testing.T) {
+	t.Parallel()
 
-	result := fs.Call(input)
-	if result != "b38699013d79e50d9d122433753959c1" {
-		t.Errorf("error executing function stack '%s': expected result "+
-			"'b38699013d79e50d9d122433753959c1' but got '%s' (%v)\n", input, result, fs)
+	type want struct {
+		arg string
+		// result of calling the returned function stack on the original input
+		callResult string
 	}
 
-	input = "maskdigits(md5sum(maskdigits($line)))"
-	fs, arg, err = NewFunctionStack(input)
-	if err != nil {
-		t.Errorf("error parsing function input '%s': %s (%v)\n", input, err.Error(), fs)
-	}
-	if arg != "$line" {
-		t.Errorf("error parsing function input '%s': expected argument '$line' but "+
-			"got '%s' (%v)\n", input, arg, fs)
+	cases := []struct {
+		input string
+		want  want
+	}{
+		{
+			input: "md5sum($line)",
+			want:  want{arg: "$line", callResult: "b38699013d79e50d9d122433753959c1"},
+		},
+		{
+			input: "maskdigits(md5sum(maskdigits($line)))",
+			want:  want{arg: "$line", callResult: ".fac.bbe..bb.........d...a.c..b."},
+		},
+		{
+			// An argument containing nested parens that are balanced is valid.
+			input: "md5sum($foo)",
+			want:  want{arg: "$foo"},
+		},
+		{
+			// Plain field with no function wrapper is a degenerate stack (empty).
+			input: "$line",
+			want:  want{arg: "$line"},
+		},
 	}
-	t.Log(input, fs, arg)
 
-	result = fs.Call(input)
-	if result != ".fac.bbe..bb.........d...a.c..b." {
-		t.Errorf("error executing function stack '%s': expected result "+
-			"'.fac.bbe..bb.........d...a.c..b.' but got '%s' (%v)\n", input, result, fs)
+	for _, tc := range cases {
+		tc := tc
+		t.Run(tc.input, func(t *testing.T) {
+			t.Parallel()
+			fs, arg, err := NewFunctionStack(tc.input)
+			if err != nil {
+				t.Fatalf("unexpected error for input %q: %v (stack %v)", tc.input, err, fs)
+			}
+			if arg != tc.want.arg {
+				t.Errorf("arg: got %q, want %q", arg, tc.want.arg)
+			}
+			if tc.want.callResult != "" {
+				got := fs.Call(tc.input)
+				if got != tc.want.callResult {
+					t.Errorf("Call(%q) = %q, want %q", tc.input, got, tc.want.callResult)
+				}
+			}
+		})
 	}
+}
+
+// TestFunctionStackMalformed verifies that NewFunctionStack rejects expressions
+// that are structurally invalid. Before the fix, several of these were silently
+// accepted and produced wrong results.
+func TestFunctionStackMalformed(t *testing.T) {
+	t.Parallel()
 
-	input = "md5sum$line)"
-	if fs, _, err := NewFunctionStack(input); err == nil {
-		t.Errorf("Expected error parsing function input '%s' (%v) but got no error\n",
-			input, fs)
+	cases := []string{
+		// Missing opening paren — no function call syntax at all.
+		"md5sum$line)",
+		// Known outer function but inner call is missing its closing paren.
+		"md5sum(makedigits$line))",
+		// Stray ')' inside the argument after stripping: "bar)baz" remains.
+		// Before the fix this was silently accepted and produced wrong output.
+		"md5sum(bar)baz)",
+		// Input ends with '(' — no closing ')' so the loop never strips,
+		// but the argument string itself contains an unclosed '('.
+		// Before the fix this was accepted as a plain field literal.
+		"foo(",
+		// Empty outer call — the name portion is empty (index == 0) which
+		// is caught by the existing index <= 0 guard.
+		"()",
 	}
 
-	input = "md5sum(makedigits$line))"
-	if fs, _, err := NewFunctionStack(input); err == nil {
-		t.Errorf("Expected error parsing function input '%s' (%v) but got no error\n",
-			input, fs)
+	for _, input := range cases {
+		input := input
+		t.Run(input, func(t *testing.T) {
+			t.Parallel()
+			fs, _, err := NewFunctionStack(input)
+			if err == nil {
+				t.Errorf("expected error for malformed input %q but got none (stack %v)", input, fs)
+			}
+		})
 	}
 }