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, 69 insertions, 0 deletions

diff --git a/internal/mapr/logformat/parser_test.go b/internal/mapr/logformat/parser_test.go
new file mode 100644
index 0000000..3b319c4
--- /dev/null
+++ b/internal/mapr/logformat/parser_test.go
@@ -0,0 +1,69 @@
+package logformat
+
+import (
+	"strings"
+	"testing"
+)
+
+type testParser struct{}
+
+func (p *testParser) MakeFields(maprLine, _ string) (map[string]string, error) {
+	return map[string]string{"line": maprLine}, nil
+}
+
+func TestRegisterParserValidation(t *testing.T) {
+	if err := RegisterParser("", wrapParserFactory(newDefaultParser)); err == nil {
+		t.Errorf("Expected error when registering parser with empty name")
+	}
+
+	if err := RegisterParser("test-nil-factory", nil); err == nil {
+		t.Errorf("Expected error when registering parser with nil factory")
+	}
+}
+
+func TestNewParserUsesRegistry(t *testing.T) {
+	const parserName = "unit-test-registry-parser"
+
+	if err := RegisterParser(parserName, func(string, string, int) (Parser, error) {
+		return &testParser{}, nil
+	}); err != nil {
+		t.Fatalf("Unable to register parser: %s", err.Error())
+	}
+
+	parser, err := NewParser(parserName, nil)
+	if err != nil {
+		t.Fatalf("Unable to create parser from registry: %s", err.Error())
+	}
+
+	fields, err := parser.MakeFields("hello", "")
+	if err != nil {
+		t.Fatalf("Unable to parse line: %s", err.Error())
+	}
+	if fields["line"] != "hello" {
+		t.Errorf("Expected custom parser output, got '%s'", fields["line"])
+	}
+}
+
+func TestNewParserFallbackToDefault(t *testing.T) {
+	parser, err := NewParser("missing-parser-format", nil)
+	if err == nil {
+		t.Fatalf("Expected NewParser to return error for missing parser format")
+	}
+	if !strings.Contains(err.Error(), "No 'missing-parser-format' mapr log format") {
+		t.Errorf("Unexpected error message: %s", err.Error())
+	}
+	if parser == nil {
+		t.Fatalf("Expected default parser fallback when format is missing")
+	}
+
+	fields, parseErr := parser.MakeFields(
+		"INFO|20211002-072342|1|parser_test.go:0|8|14|7|0.21|471h0m21s|MAPREDUCE:STATS|foo=bar",
+		"",
+	)
+	if parseErr != nil {
+		t.Fatalf("Fallback parser failed to parse line: %s", parseErr.Error())
+	}
+	if val, ok := fields["$severity"]; !ok || val != "INFO" {
+		t.Errorf("Fallback parser did not behave like default parser")
+	}
+}