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

diff --git a/internal/discovery/discovery_test.go b/internal/discovery/discovery_test.go
new file mode 100644
index 0000000..afc5d31
--- /dev/null
+++ b/internal/discovery/discovery_test.go
@@ -0,0 +1,176 @@
+package discovery
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/mimecast/dtail/internal/io/dlog"
+)
+
+func TestMain(m *testing.M) {
+	dlog.Common = &dlog.DLog{}
+	m.Run()
+}
+
+// TestShuffleListDoesNotMutateInput verifies that shuffleList never writes
+// back into the caller's backing array, and that repeated calls within the
+// same sub-second window produce different orderings (the chance of a false
+// failure with 20 elements is 1/20! ≈ 4×10⁻¹⁹).
+func TestShuffleListDoesNotMutateInput(t *testing.T) {
+	t.Parallel()
+
+	d := &Discovery{}
+
+	// Build a stable reference slice.
+	original := make([]string, 20)
+	for i := range original {
+		original[i] = string(rune('a' + i))
+	}
+
+	snapshot := append([]string(nil), original...)
+
+	// First shuffle — must not touch original.
+	first := d.shuffleList(original)
+	if !reflect.DeepEqual(original, snapshot) {
+		t.Fatalf("shuffleList mutated caller's slice: got %v, want %v", original, snapshot)
+	}
+	if len(first) != len(original) {
+		t.Fatalf("shuffleList returned wrong length: got %d, want %d", len(first), len(original))
+	}
+
+	// Second shuffle — different seed path; overwhelmingly likely to differ.
+	second := d.shuffleList(original)
+	if reflect.DeepEqual(first, second) {
+		t.Fatal("shuffleList returned identical order on two consecutive calls; seed resolution may be too coarse")
+	}
+
+	// Both results must contain exactly the same elements as the input.
+	countFirst := make(map[string]int, len(original))
+	countSecond := make(map[string]int, len(original))
+	for i := range original {
+		countFirst[first[i]]++
+		countSecond[second[i]]++
+	}
+	for _, s := range original {
+		if countFirst[s] != 1 {
+			t.Errorf("element %q appears %d times in first shuffle, want 1", s, countFirst[s])
+		}
+		if countSecond[s] != 1 {
+			t.Errorf("element %q appears %d times in second shuffle, want 1", s, countSecond[s])
+		}
+	}
+}
+
+func TestNewParsesModuleOptionsWithAdditionalColons(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name    string
+		method  string
+		wantMod string
+		wantOpt string
+		wantErr bool
+	}{
+		{
+			name:    "plain module",
+			method:  "file",
+			wantMod: "FILE",
+		},
+		{
+			name:    "options with additional colons",
+			method:  "method:host:port:extra",
+			wantMod: "METHOD",
+			wantOpt: "host:port:extra",
+		},
+		{
+			name:    "empty options rejected",
+			method:  "method:",
+			wantErr: true,
+		},
+		{
+			name:    "missing module rejected",
+			method:  ":host:port",
+			wantErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			got, err := New(tt.method, "server", Shuffle)
+			if tt.wantErr {
+				if err == nil {
+					t.Fatalf("New(%q) error = nil, want error", tt.method)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("New(%q) error = %v, want nil", tt.method, err)
+			}
+			if got.module != tt.wantMod {
+				t.Fatalf("module = %q, want %q", got.module, tt.wantMod)
+			}
+			if got.options != tt.wantOpt {
+				t.Fatalf("options = %q, want %q", got.options, tt.wantOpt)
+			}
+		})
+	}
+}
+
+func TestServerListIgnoresEmptyServerEntries(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name    string
+		method  string
+		server  string
+		want    []string
+		wantErr bool
+	}{
+		{
+			name:   "regex without module yields no phantom host",
+			server: "/.*/",
+			want:   []string{},
+		},
+		{
+			name:   "comma list filters empty entries",
+			server: "alpha,,beta,",
+			want:   []string{"alpha", "beta"},
+		},
+		{
+			name:   "empty server input preserves serverless sentinel",
+			server: "",
+			want:   []string{""},
+		},
+	}
+
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			got, err := New(tt.method, tt.server, ServerOrder(99))
+			if tt.wantErr {
+				if err == nil {
+					t.Fatalf("New(%q, %q) error = nil, want error", tt.method, tt.server)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("New(%q, %q) error = %v, want nil", tt.method, tt.server, err)
+			}
+
+			servers := got.ServerList()
+			if len(servers) != len(tt.want) {
+				t.Fatalf("ServerList() len = %d, want %d (%v)", len(servers), len(tt.want), servers)
+			}
+			for i := range tt.want {
+				if servers[i] != tt.want[i] {
+					t.Fatalf("ServerList()[%d] = %q, want %q (full=%v)", i, servers[i], tt.want[i], servers)
+				}
+			}
+		})
+	}
+}