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>

3 files changed, 210 insertions, 15 deletions

diff --git a/internal/discovery/comma.go b/internal/discovery/comma.go
index 9bea89c..126cbec 100644
--- a/internal/discovery/comma.go
+++ b/internal/discovery/comma.go
@@ -9,5 +9,18 @@ import (
 // ServerListFromCOMMA retrieves a list of servers from comma separated input list.
 func (d *Discovery) ServerListFromCOMMA() []string {
 	dlog.Common.Debug("Retrieving server list from comma separated list", d.server)
-	return strings.Split(d.server, ",")
+
+	rawServers := strings.Split(d.server, ",")
+	servers := make([]string, 0, len(rawServers))
+	for _, server := range rawServers {
+		if server == "" {
+			continue
+		}
+		servers = append(servers, server)
+	}
+	if len(servers) == 0 && d.server == "" {
+		return rawServers
+	}
+
+	return servers
 }
diff --git a/internal/discovery/discovery.go b/internal/discovery/discovery.go
index f709501..bebd798 100644
--- a/internal/discovery/discovery.go
+++ b/internal/discovery/discovery.go
@@ -35,14 +35,14 @@ type Discovery struct {
 }
 
 // New returns a new discovery method.
-func New(method, server string, order ServerOrder) *Discovery {
+func New(method, server string, order ServerOrder) (*Discovery, error) {
 	module := method
 	options := ""
 
 	if strings.Contains(module, ":") {
-		s := strings.Split(module, ":")
-		if len(s) != 2 {
-			dlog.Common.FatalPanic("Unable to parse discovery module", module)
+		s := strings.SplitN(module, ":", 2)
+		if len(s) != 2 || s[0] == "" || s[1] == "" {
+			return nil, fmt.Errorf("unable to parse discovery module %q", module)
 		}
 		module = s[0]
 		options = s[1]
@@ -59,7 +59,7 @@ func New(method, server string, order ServerOrder) *Discovery {
 		d.initRegex()
 	}
 
-	return &d
+	return &d, nil
 }
 
 func (d *Discovery) initRegex() {
@@ -102,6 +102,9 @@ func (d *Discovery) serverListFromModule() []string {
 	if d.module != "" {
 		return d.serverListFromReflectedModule()
 	}
+	if d.server == "" && d.regex != nil {
+		return []string{}
+	}
 	if _, err := os.Stat(d.server); err == nil {
 		// Appears to be a file name, now try to read from that file.
 		return d.ServerListFromFILE()
@@ -153,19 +156,22 @@ func (d *Discovery) dedupList(servers []string) (deduped []string) {
 	return
 }
 
-// Randomly shuffle the server list.
+// Randomly shuffle the server list. A copy of the input slice is made so the
+// caller's backing array is never modified. The seed uses UnixNano so that
+// two callers started within the same wall-clock second still get different
+// shuffle orders, which is important for client-side load spreading.
 func (d *Discovery) shuffleList(servers []string) []string {
 	dlog.Common.Debug("Shuffling server list")
 
-	r := rand.New(rand.NewSource(time.Now().Unix()))
-	shuffled := make([]string, len(servers))
-	n := len(servers)
+	// Copy to avoid mutating the caller's backing array.
+	shuffled := append([]string(nil), servers...)
 
-	for i := 0; i < n; i++ {
-		randIndex := r.Intn(len(servers))
-		shuffled[i] = servers[randIndex]
-		servers = append(servers[:randIndex], servers[randIndex+1:]...)
-	}
+	// Seed with nanosecond resolution to avoid identical orderings when
+	// multiple clients start within the same second.
+	r := rand.New(rand.NewSource(time.Now().UnixNano()))
+	r.Shuffle(len(shuffled), func(i, j int) {
+		shuffled[i], shuffled[j] = shuffled[j], shuffled[i]
+	})
 
 	return shuffled
 }
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)
+				}
+			}
+		})
+	}
+}