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, 20 insertions, 14 deletions

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
 }