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, 12 insertions, 7 deletions

diff --git a/internal/server/scheduler.go b/internal/server/scheduler.go
index cc51b7e..fa47d2e 100644
--- a/internal/server/scheduler.go
+++ b/internal/server/scheduler.go
@@ -16,10 +16,12 @@ import (
 	gossh "golang.org/x/crypto/ssh"
 )
 
-type scheduler struct{}
+type scheduler struct {
+	cfg config.RuntimeConfig
+}
 
-func newScheduler() *scheduler {
-	return &scheduler{}
+func newScheduler(cfg config.RuntimeConfig) *scheduler {
+	return &scheduler{cfg: cfg}
 }
 
 func (s *scheduler) start(ctx context.Context) {
@@ -27,9 +29,11 @@ func (s *scheduler) start(ctx context.Context) {
 	// First run after just 10s!
 	time.Sleep(time.Second * 2)
 	s.runJobs(ctx)
+	runTicker := time.NewTicker(time.Minute)
+	defer runTicker.Stop()
 	for {
 		select {
-		case <-time.After(time.Minute):
+		case <-runTicker.C:
 			s.runJobs(ctx)
 		case <-ctx.Done():
 			return
@@ -38,7 +42,8 @@ func (s *scheduler) start(ctx context.Context) {
 }
 
 func (s *scheduler) runJobs(ctx context.Context) {
-	for _, job := range config.Server.Schedule {
+	for i := range s.cfg.Server.Schedule {
+		job := &s.cfg.Server.Schedule[i]
 		if !job.Enable {
 			dlog.Server.Debug(job.Name, "Not running job as not enabled")
 			continue
@@ -56,7 +61,7 @@ func (s *scheduler) runJobs(ctx context.Context) {
 	}
 }
 
-func (s *scheduler) runJob(ctx context.Context, job config.Scheduled) {
+func (s *scheduler) runJob(ctx context.Context, job *config.Scheduled) {
 	files := fillDates(job.Files)
 	outfile := fillDates(job.Outfile)
 
@@ -68,7 +73,7 @@ func (s *scheduler) runJob(ctx context.Context, job config.Scheduled) {
 
 	servers := strings.Join(job.Servers, ",")
 	if servers == "" {
-		servers = config.Server.SSHBindAddress
+		servers = s.cfg.Server.SSHBindAddress
 	}
 	args := config.Args{
 		ConnectionsPerCPU: config.DefaultConnectionsPerCPU,