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>

2 files changed, 128 insertions, 50 deletions

diff --git a/internal/mapr/funcs/function.go b/internal/mapr/funcs/function.go
index 418d86f..2f21d5a 100644
--- a/internal/mapr/funcs/function.go
+++ b/internal/mapr/funcs/function.go
@@ -20,20 +20,10 @@ type Function struct {
 type FunctionStack []Function
 
 // NewFunctionStack parses the input string, e.g. foo(bar("arg")) and returns
-// a corresponding function stack.
+// a corresponding function stack. It returns an error for malformed inputs
+// such as unbalanced parentheses (e.g. "foo(", "foo(bar)baz").
 func NewFunctionStack(in string) (FunctionStack, string, error) {
 	var fs FunctionStack
-	getCallback := func(name string) (CallbackFunc, error) {
-		var cb CallbackFunc
-		switch name {
-		case "md5sum":
-			return Md5Sum, nil
-		case "maskdigits":
-			return MaskDigits, nil
-		default:
-			return cb, fmt.Errorf("unknown function '%s'", name)
-		}
-	}
 
 	aux := in
 	for strings.HasSuffix(aux, ")") {
@@ -43,16 +33,64 @@ func NewFunctionStack(in string) (FunctionStack, string, error) {
 		}
 		name := aux[0:index]
 
-		call, err := getCallback(name)
+		call, err := lookupCallback(name)
 		if err != nil {
 			return fs, "", err
 		}
 		fs = append(fs, Function{name, call})
+		// Strip the outer function name and its enclosing parens, leaving
+		// only the argument expression for the next iteration.
 		aux = aux[index+1 : len(aux)-1]
 	}
+
+	// Validate that no unbalanced parentheses remain in the argument string.
+	// Inputs like "foo(bar)baz" leave "bar)baz" after stripping, and inputs
+	// ending with "(" (no closing ")") are accepted as plain field literals
+	// without this check — both produce silently wrong behavior.
+	if err := validateParenBalance(aux, in); err != nil {
+		return fs, "", err
+	}
+
 	return fs, aux, nil
 }
 
+// lookupCallback maps a function name to its CallbackFunc implementation.
+// It returns an error for unrecognised names so callers get a clear message.
+func lookupCallback(name string) (CallbackFunc, error) {
+	switch name {
+	case "md5sum":
+		return Md5Sum, nil
+	case "maskdigits":
+		return MaskDigits, nil
+	default:
+		var zero CallbackFunc
+		return zero, fmt.Errorf("unknown function '%s'", name)
+	}
+}
+
+// validateParenBalance checks that the remaining argument string contains no
+// unbalanced parentheses. A negative depth means a stray ')' was found; a
+// non-zero depth after the loop means an unclosed '(' was found. The original
+// full expression is included in the error message for context.
+func validateParenBalance(aux, original string) error {
+	depth := 0
+	for _, r := range aux {
+		switch r {
+		case '(':
+			depth++
+		case ')':
+			depth--
+		}
+		if depth < 0 {
+			return fmt.Errorf("malformed function expression %q: unexpected ')' in argument", original)
+		}
+	}
+	if depth != 0 {
+		return fmt.Errorf("malformed function expression %q: unclosed '(' in argument", original)
+	}
+	return nil
+}
+
 // Call the function stack.
 func (fs FunctionStack) Call(str string) string {
 	for i := len(fs) - 1; i >= 0; i-- {
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)
+			}
+		})
 	}
 }