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>

6 files changed, 472 insertions, 0 deletions

diff --git a/internal/cli/authkeyflags.go b/internal/cli/authkeyflags.go
new file mode 100644
index 0000000..4c945d3
--- /dev/null
+++ b/internal/cli/authkeyflags.go
@@ -0,0 +1,43 @@
+package cli
+
+import (
+	"flag"
+
+	"github.com/mimecast/dtail/internal/config"
+)
+
+const authKeyPathHelpText = "Path to auth key/private key (defaults to ~/.ssh/id_rsa via config)"
+
+// BindAuthKeyFlags registers the legacy and current auth-key flags.
+func BindAuthKeyFlags(fs *flag.FlagSet, legacyKey *string, args *config.Args) {
+	fs.StringVar(legacyKey, "key", "", "Deprecated alias for -auth-key-path")
+	fs.StringVar(&args.SSHPrivateKeyFilePath, "auth-key-path", "", authKeyPathHelpText)
+}
+
+// FlagWasSet reports whether the named flag was explicitly set on the command line.
+func FlagWasSet(name string) bool {
+	var wasSet bool
+	flag.Visit(func(f *flag.Flag) {
+		if f.Name == name {
+			wasSet = true
+		}
+	})
+	return wasSet
+}
+
+// ApplyAuthKeyPathCompatibility copies the deprecated legacy key into args when
+// the new flag was not explicitly set. If both were set, the new flag wins and
+// a warning message is returned.
+func ApplyAuthKeyPathCompatibility(args *config.Args, legacyKey string, authKeyPathSet bool) string {
+	if authKeyPathSet {
+		if legacyKey != "" {
+			return "WARN: -key is deprecated; ignoring it because -auth-key-path was also set"
+		}
+		return ""
+	}
+
+	if legacyKey != "" {
+		args.SSHPrivateKeyFilePath = legacyKey
+	}
+	return ""
+}
diff --git a/internal/cli/authkeyflags_test.go b/internal/cli/authkeyflags_test.go
new file mode 100644
index 0000000..64c76c3
--- /dev/null
+++ b/internal/cli/authkeyflags_test.go
@@ -0,0 +1,94 @@
+package cli
+
+import (
+	"flag"
+	"testing"
+
+	"github.com/mimecast/dtail/internal/config"
+)
+
+func TestApplyAuthKeyPathCompatibilityLegacyOnly(t *testing.T) {
+	args := config.Args{}
+
+	if warning := ApplyAuthKeyPathCompatibility(&args, "/tmp/legacy.pem", false); warning != "" {
+		t.Fatalf("unexpected warning: %q", warning)
+	}
+
+	if got, want := args.SSHPrivateKeyFilePath, "/tmp/legacy.pem"; got != want {
+		t.Fatalf("unexpected auth key path: want %q got %q", want, got)
+	}
+}
+
+func TestApplyAuthKeyPathCompatibilityPrefersExplicitAuthKeyPath(t *testing.T) {
+	tests := []struct {
+		name string
+		args []string
+		want string
+		warn bool
+	}{
+		{
+			name: "legacy then auth",
+			args: []string{"-key", "/tmp/legacy.pem", "-auth-key-path", "/tmp/current.pem"},
+			want: "/tmp/current.pem",
+			warn: true,
+		},
+		{
+			name: "auth then legacy",
+			args: []string{"-auth-key-path", "/tmp/current.pem", "-key", "/tmp/legacy.pem"},
+			want: "/tmp/current.pem",
+			warn: true,
+		},
+		{
+			name: "explicit blank auth key path keeps blank",
+			args: []string{"-key", "/tmp/legacy.pem", "-auth-key-path="},
+			want: "",
+			warn: true,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			fs := flag.NewFlagSet(t.Name(), flag.ContinueOnError)
+			var args config.Args
+			var legacyKey string
+
+			BindAuthKeyFlags(fs, &legacyKey, &args)
+			if err := fs.Parse(tc.args); err != nil {
+				t.Fatalf("Parse failed: %v", err)
+			}
+
+			var authKeyPathSet bool
+			fs.Visit(func(f *flag.Flag) {
+				if f.Name == "auth-key-path" {
+					authKeyPathSet = true
+				}
+			})
+
+			warning := ApplyAuthKeyPathCompatibility(&args, legacyKey, authKeyPathSet)
+			if gotWarn := warning != ""; gotWarn != tc.warn {
+				t.Fatalf("unexpected warning presence: want %v got %v (%q)", tc.warn, gotWarn, warning)
+			}
+			if got, want := args.SSHPrivateKeyFilePath, tc.want; got != want {
+				t.Fatalf("unexpected auth key path: want %q got %q", want, got)
+			}
+		})
+	}
+}
+
+func TestBindAuthKeyFlagsHelpText(t *testing.T) {
+	fs := flag.NewFlagSet(t.Name(), flag.ContinueOnError)
+	var args config.Args
+	var legacyKey string
+
+	BindAuthKeyFlags(fs, &legacyKey, &args)
+
+	if got, want := fs.Lookup("key").Usage, "Deprecated alias for -auth-key-path"; got != want {
+		t.Fatalf("unexpected legacy flag help: want %q got %q", want, got)
+	}
+	if got, want := fs.Lookup("auth-key-path").Usage, authKeyPathHelpText; got != want {
+		t.Fatalf("unexpected auth-key-path help: want %q got %q", want, got)
+	}
+	if got, want := fs.Lookup("auth-key-path").DefValue, ""; got != want {
+		t.Fatalf("unexpected auth-key-path default: want %q got %q", want, got)
+	}
+}
diff --git a/internal/cli/pprof.go b/internal/cli/pprof.go
new file mode 100644
index 0000000..57cb38c
--- /dev/null
+++ b/internal/cli/pprof.go
@@ -0,0 +1,119 @@
+package cli
+
+import (
+	"context"
+	"errors"
+	"net"
+	"net/http"
+	"net/http/pprof"
+	"runtime"
+	"sync"
+
+	"github.com/mimecast/dtail/internal/io/dlog"
+)
+
+// Mutex and block profiling rates. The /debug/pprof/mutex and
+// /debug/pprof/block endpoints return empty profiles unless these runtime
+// rates are enabled first; they are off by default in the Go runtime.
+//
+// The values below were validated during task ss0: a mutex profile fraction of
+// 5 (report ~1/5 of contention events) and a block profile rate of 100000ns
+// (record blocking events longer than 100µs) were sufficient to measure
+// outputManager mutex contention — 12.4ms total delay across three concurrent
+// 100MB server-mode dcats, i.e. NOT a bottleneck; an idle follow session
+// costs 0 CPU ticks/10s because the 1ms read poll only runs while direct output is
+// active and the EOF-ack drops it back to 1s polling. Documented here so the
+// locking design is not re-litigated.
+const (
+	mutexProfileFraction  = 5
+	blockProfileRateNanos = 100000
+)
+
+// PProfServer owns a dedicated pprof HTTP server lifecycle.
+type PProfServer struct {
+	listener net.Listener
+	server   *http.Server
+	done     chan struct{}
+}
+
+// NewPProfServer creates a pprof HTTP server bound to address.
+func NewPProfServer(address string) (*PProfServer, error) {
+	listener, err := net.Listen("tcp", address)
+	if err != nil {
+		return nil, err
+	}
+
+	return &PProfServer{
+		listener: listener,
+		server: &http.Server{
+			Handler: newPProfServeMux(),
+		},
+		done: make(chan struct{}),
+	}, nil
+}
+
+// EnableProfilingRates turns on mutex and block profiling collection so that
+// the /debug/pprof/mutex and /debug/pprof/block endpoints actually contain
+// samples. Without this the Go runtime keeps both rates at zero and those
+// endpoints report empty profiles. Only call this when pprof is enabled so it
+// costs nothing in the common (no --pprof) case.
+func EnableProfilingRates() {
+	runtime.SetMutexProfileFraction(mutexProfileFraction)
+	runtime.SetBlockProfileRate(blockProfileRateNanos)
+}
+
+// Address returns the bound pprof listener address.
+func (s *PProfServer) Address() string {
+	if s == nil || s.listener == nil {
+		return ""
+	}
+	return s.listener.Addr().String()
+}
+
+// Start serves the pprof HTTP endpoints until shutdown.
+func (s *PProfServer) Start(wg *sync.WaitGroup) {
+	if s == nil {
+		return
+	}
+
+	if wg != nil {
+		wg.Add(1)
+	}
+
+	go func() {
+		if wg != nil {
+			defer wg.Done()
+		}
+		defer close(s.done)
+
+		if err := s.server.Serve(s.listener); err != nil && !errors.Is(err, http.ErrServerClosed) {
+			dlog.Client.Error("PProf server exited", err)
+		}
+	}()
+}
+
+// Shutdown stops the pprof HTTP server and waits for Serve to return.
+func (s *PProfServer) Shutdown(ctx context.Context) error {
+	if s == nil {
+		return nil
+	}
+
+	err := s.server.Shutdown(ctx)
+	<-s.done
+	return err
+}
+
+func newPProfServeMux() *http.ServeMux {
+	mux := http.NewServeMux()
+	mux.HandleFunc("/debug/pprof/", pprof.Index)
+	mux.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)
+	mux.HandleFunc("/debug/pprof/profile", pprof.Profile)
+	mux.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
+	mux.HandleFunc("/debug/pprof/trace", pprof.Trace)
+
+	for _, name := range []string{"allocs", "block", "goroutine", "heap", "mutex", "threadcreate"} {
+		mux.Handle("/debug/pprof/"+name, pprof.Handler(name))
+	}
+
+	return mux
+}
diff --git a/internal/cli/pprof_test.go b/internal/cli/pprof_test.go
new file mode 100644
index 0000000..aa1cd05
--- /dev/null
+++ b/internal/cli/pprof_test.go
@@ -0,0 +1,41 @@
+package cli
+
+import (
+	"runtime"
+	"testing"
+)
+
+// TestEnableProfilingRatesSetsMutexFraction verifies that EnableProfilingRates
+// actually turns on mutex profiling. Without this the /debug/pprof/mutex
+// endpoint reports an empty profile. runtime.SetMutexProfileFraction(-1) reads
+// the current fraction without changing it, so it lets us assert the state.
+func TestEnableProfilingRatesSetsMutexFraction(t *testing.T) {
+	// Save and restore global runtime state so this test does not leak into
+	// other tests in the package. The block profile rate has no getter, so we
+	// simply disable it again on cleanup.
+	prevMutex := runtime.SetMutexProfileFraction(-1)
+	t.Cleanup(func() {
+		runtime.SetMutexProfileFraction(prevMutex)
+		runtime.SetBlockProfileRate(0)
+	})
+
+	// Start from a known-disabled state.
+	runtime.SetMutexProfileFraction(0)
+
+	EnableProfilingRates()
+
+	if got := runtime.SetMutexProfileFraction(-1); got != mutexProfileFraction {
+		t.Errorf("mutex profile fraction = %d, want %d", got, mutexProfileFraction)
+	}
+}
+
+// TestProfilingRateConstants pins the ss0-validated values so a change is a
+// conscious decision rather than an accident.
+func TestProfilingRateConstants(t *testing.T) {
+	if mutexProfileFraction != 5 {
+		t.Errorf("mutexProfileFraction = %d, want 5", mutexProfileFraction)
+	}
+	if blockProfileRateNanos != 100000 {
+		t.Errorf("blockProfileRateNanos = %d, want 100000", blockProfileRateNanos)
+	}
+}
diff --git a/internal/cli/runtime.go b/internal/cli/runtime.go
new file mode 100644
index 0000000..950ed33
--- /dev/null
+++ b/internal/cli/runtime.go
@@ -0,0 +1,104 @@
+package cli
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/mimecast/dtail/internal/io/dlog"
+	"github.com/mimecast/dtail/internal/profiling"
+	"github.com/mimecast/dtail/internal/source"
+)
+
+// ClientRuntime owns common client command runtime components.
+type ClientRuntime struct {
+	ctx            context.Context
+	cancel         context.CancelFunc
+	wg             sync.WaitGroup
+	pprofServer    *PProfServer
+	profiler       *profiling.Profiler
+	profileEnabled bool
+}
+
+// NewClientRuntime starts logging and profiling for a client command.
+func NewClientRuntime(parent context.Context, profileFlags profiling.Flags, profileName string) *ClientRuntime {
+	if parent == nil {
+		parent = context.Background()
+	}
+	ctx, cancel := context.WithCancel(parent)
+	runtime := &ClientRuntime{
+		ctx:            ctx,
+		cancel:         cancel,
+		profiler:       profiling.NewProfiler(profileFlags.ToConfig(profileName)),
+		profileEnabled: profileFlags.Enabled(),
+	}
+
+	runtime.wg.Add(1)
+	dlog.Start(ctx, &runtime.wg, source.Client)
+	return runtime
+}
+
+// Context returns the runtime context.
+func (r *ClientRuntime) Context() context.Context {
+	return r.ctx
+}
+
+// Cancel cancels the runtime context.
+func (r *ClientRuntime) Cancel() {
+	r.cancel()
+}
+
+// StartPProf starts the pprof server if an address is provided.
+func (r *ClientRuntime) StartPProf(address string) {
+	if address == "" {
+		return
+	}
+
+	r.stopPProf()
+
+	server, err := NewPProfServer(address)
+	if err != nil {
+		dlog.Client.Error("Unable to start PProf", err)
+		return
+	}
+
+	r.pprofServer = server
+	dlog.Client.Info("Starting PProf", server.Address())
+	server.Start(&r.wg)
+}
+
+// LogStartupMetrics logs startup profiling metrics when enabled.
+func (r *ClientRuntime) LogStartupMetrics() {
+	if r.profileEnabled {
+		r.profiler.LogMetrics("startup")
+	}
+}
+
+// LogShutdownMetrics logs shutdown profiling metrics when enabled.
+func (r *ClientRuntime) LogShutdownMetrics() {
+	if r.profileEnabled {
+		r.profiler.LogMetrics("shutdown")
+	}
+}
+
+// Stop stops profiling and logging runtime goroutines.
+func (r *ClientRuntime) Stop() {
+	r.profiler.Stop()
+	r.stopPProf()
+	r.cancel()
+	r.wg.Wait()
+}
+
+func (r *ClientRuntime) stopPProf() {
+	if r.pprofServer == nil {
+		return
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	if err := r.pprofServer.Shutdown(ctx); err != nil {
+		dlog.Client.Error("Unable to stop PProf", err)
+	}
+	r.pprofServer = nil
+}
diff --git a/internal/cli/runtime_test.go b/internal/cli/runtime_test.go
new file mode 100644
index 0000000..e024a4e
--- /dev/null
+++ b/internal/cli/runtime_test.go
@@ -0,0 +1,71 @@
+package cli
+
+import (
+	"context"
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/mimecast/dtail/internal/io/dlog"
+	"github.com/mimecast/dtail/internal/profiling"
+)
+
+func TestClientRuntimeStopShutsDownPProf(t *testing.T) {
+	prevClient := dlog.Client
+	dlog.Client = &dlog.DLog{}
+	t.Cleanup(func() {
+		dlog.Client = prevClient
+	})
+
+	ctx, cancel := context.WithCancel(context.Background())
+	runtime := &ClientRuntime{
+		ctx:      ctx,
+		cancel:   cancel,
+		profiler: profiling.NewProfiler(profiling.Config{}),
+	}
+
+	runtime.StartPProf("127.0.0.1:0")
+	if runtime.pprofServer == nil {
+		t.Fatal("expected pprof server to start")
+	}
+
+	url := "http://" + runtime.pprofServer.Address() + "/debug/pprof/"
+	waitForHTTPStatus(t, url, http.StatusOK)
+
+	runtime.Stop()
+	waitForHTTPError(t, url)
+}
+
+func waitForHTTPStatus(t *testing.T, url string, want int) {
+	t.Helper()
+
+	deadline := time.Now().Add(3 * time.Second)
+	for time.Now().Before(deadline) {
+		resp, err := http.Get(url)
+		if err == nil {
+			resp.Body.Close()
+			if resp.StatusCode == want {
+				return
+			}
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+
+	t.Fatalf("timed out waiting for %s to return %d", url, want)
+}
+
+func waitForHTTPError(t *testing.T, url string) {
+	t.Helper()
+
+	deadline := time.Now().Add(3 * time.Second)
+	for time.Now().Before(deadline) {
+		resp, err := http.Get(url)
+		if err != nil {
+			return
+		}
+		resp.Body.Close()
+		time.Sleep(10 * time.Millisecond)
+	}
+
+	t.Fatalf("timed out waiting for %s to stop serving", url)
+}