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, 17 insertions, 11 deletions

diff --git a/internal/clients/stats.go b/internal/clients/stats.go
index 7a6643b..1ce04e6 100644
--- a/internal/clients/stats.go
+++ b/internal/clients/stats.go
@@ -8,8 +8,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/mimecast/dtail/internal/color"
-	"github.com/mimecast/dtail/internal/config"
 	"github.com/mimecast/dtail/internal/io/dlog"
 	"github.com/mimecast/dtail/internal/protocol"
 )
@@ -24,13 +22,22 @@ type stats struct {
 	connected int
 	// To synchronize concurrent access.
 	mutex sync.Mutex
+	// Formats interrupt-driven stats output.
+	formatter interruptMessageFormatter
+	// Controls how long interrupt output remains visible.
+	interruptPause time.Duration
 }
 
-func newTailStats(servers int) *stats {
+func newTailStats(servers int, formatter interruptMessageFormatter, interruptPause time.Duration) *stats {
+	if interruptPause <= 0 {
+		interruptPause = 3 * time.Second
+	}
 	return &stats{
 		servers:          servers,
 		connectionsEstCh: make(chan struct{}, servers),
 		connected:        0,
+		formatter:        formatter,
+		interruptPause:   interruptPause,
 	}
 }
 
@@ -39,6 +46,9 @@ func newTailStats(servers int) *stats {
 func (s *stats) Start(ctx context.Context, throttleCh <-chan struct{},
 	statsCh <-chan string, quiet bool) {
 
+	ticker := time.NewTicker(3 * time.Second)
+	defer ticker.Stop()
+
 	var connectedLast int
 	for {
 		var force bool
@@ -48,7 +58,7 @@ func (s *stats) Start(ctx context.Context, throttleCh <-chan struct{},
 		case message := <-statsCh:
 			messages = append(messages, message)
 			force = true
-		case <-time.After(time.Second * 3):
+		case <-ticker.C:
 		case <-ctx.Done():
 			return
 		}
@@ -81,17 +91,13 @@ func (s *stats) Start(ctx context.Context, throttleCh <-chan struct{},
 func (s *stats) printStatsDueInterrupt(messages []string) {
 	dlog.Client.Pause()
 	for i, message := range messages {
-		if i > 0 && config.Client.TermColorsEnable {
-			fmt.Println(color.PaintStrWithAttr(message,
-				config.Client.TermColors.Client.ClientFg,
-				config.Client.TermColors.Client.ClientBg,
-				config.Client.TermColors.Client.ClientAttr,
-			))
+		if s.formatter != nil {
+			fmt.Println(s.formatter.FormatInterruptMessage(i, message))
 			continue
 		}
 		fmt.Printf(" %s\n", message)
 	}
-	time.Sleep(time.Second * time.Duration(config.InterruptTimeoutS))
+	time.Sleep(s.interruptPause)
 	dlog.Client.Resume()
 }