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, 51 insertions, 13 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-- {