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, 59 insertions, 4 deletions

diff --git a/internal/server/stats.go b/internal/server/stats.go
index 99a644a..7a60d61 100644
--- a/internal/server/stats.go
+++ b/internal/server/stats.go
@@ -6,7 +6,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/mimecast/dtail/internal/config"
 	"github.com/mimecast/dtail/internal/io/dlog"
 )
 
@@ -15,6 +14,20 @@ type stats struct {
 	mutex               sync.Mutex
 	currentConnections  int
 	lifetimeConnections uint64
+	maxConnections      int
+	// preAuthConnections counts TCP connections that have been accepted but
+	// whose SSH handshake has not yet completed. These are counted against
+	// maxConnections so that slow or abusive clients cannot create unbounded
+	// goroutines during the handshake phase. Once a handshake succeeds the
+	// slot is converted to a regular connection (decrementPreAuth +
+	// incrementConnections). On failure the slot is simply released.
+	preAuthConnections int
+}
+
+func newStats(maxConnections int) stats {
+	return stats{
+		maxConnections: maxConnections,
+	}
 }
 
 func (s *stats) incrementConnections() {
@@ -32,6 +45,40 @@ func (s *stats) decrementConnections() {
 	s.mutex.Unlock()
 }
 
+// reservePreAuth increments the pre-auth counter immediately after Accept so
+// that slow or unauthenticated handshakes are counted against maxConnections.
+// It must be paired with exactly one call to releasePreAuth or
+// promotePreAuthToConnection.
+func (s *stats) reservePreAuth() {
+	defer s.logServerStats()
+	s.mutex.Lock()
+	s.preAuthConnections++
+	s.mutex.Unlock()
+}
+
+// releasePreAuth decrements the pre-auth counter without converting the slot
+// into a full authenticated connection. Call this on every handshake failure
+// path to undo the reservePreAuth reservation.
+func (s *stats) releasePreAuth() {
+	defer s.logServerStats()
+	s.mutex.Lock()
+	s.preAuthConnections--
+	s.mutex.Unlock()
+}
+
+// promotePreAuthToConnection atomically converts a pre-auth reservation into a
+// full authenticated connection. It decrements preAuthConnections and
+// increments both currentConnections and lifetimeConnections under a single
+// lock acquisition so there is no instant where neither counter holds the slot.
+func (s *stats) promotePreAuthToConnection() {
+	defer s.logServerStats()
+	s.mutex.Lock()
+	s.preAuthConnections--
+	s.currentConnections++
+	s.lifetimeConnections++
+	s.mutex.Unlock()
+}
+
 func (s *stats) hasConnections() bool {
 	s.mutex.Lock()
 	currentConnections := s.currentConnections
@@ -50,16 +97,24 @@ func (s *stats) logServerStats() {
 	data := make(map[string]interface{})
 	data["currentConnections"] = s.currentConnections
 	data["lifetimeConnections"] = s.lifetimeConnections
+	data["preAuthConnections"] = s.preAuthConnections
 	dlog.Server.Mapreduce("STATS", data)
 }
 
+// serverLimitExceeded checks whether accepting another connection would exceed
+// maxConnections. Both authenticated connections (currentConnections) and
+// in-progress handshakes (preAuthConnections) are counted so that slow or
+// unauthenticated clients cannot bypass the limit by keeping many TCP
+// connections open during the handshake phase.
 func (s *stats) serverLimitExceeded() error {
 	s.mutex.Lock()
 	defer s.mutex.Unlock()
 
-	if s.currentConnections >= config.Server.MaxConnections {
-		return fmt.Errorf("Exceeded max allowed concurrent connections of %d",
-			config.Server.MaxConnections)
+	// Count both authenticated connections and pre-auth handshakes in progress.
+	total := s.currentConnections + s.preAuthConnections
+	if total >= s.maxConnections {
+		return fmt.Errorf("Exceeded max allowed concurrent connections of %d (current=%d, pre-auth=%d)",
+			s.maxConnections, s.currentConnections, s.preAuthConnections)
 	}
 	return nil
 }