From 849951be1d1a7ee9f9302006ccb187bf5b4e36f3 Mon Sep 17 00:00:00 2001
From: Paul Buetow <paul@buetow.org>
Date: Wed, 22 Jul 2026 23:51:18 +0300
Subject: =?UTF-8?q?feat:=20DTail=20fork=20=E2=80=94=20server/client=20feat?=
 =?UTF-8?q?ure=20development?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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>
---
 internal/mapr/safe_aggregateset.go | 72 ++++++++++++++++++++++++++++++++++++++
 1 file changed, 72 insertions(+)
 create mode 100644 internal/mapr/safe_aggregateset.go

(limited to 'internal/mapr/safe_aggregateset.go')

diff --git a/internal/mapr/safe_aggregateset.go b/internal/mapr/safe_aggregateset.go
new file mode 100644
index 0000000..5d64f3e
--- /dev/null
+++ b/internal/mapr/safe_aggregateset.go
@@ -0,0 +1,72 @@
+package mapr
+
+import (
+	"context"
+	"sync"
+)
+
+// SafeAggregateSet is a thread-safe wrapper around AggregateSet for concurrent aggregation.
+// It uses a read-write mutex to allow concurrent reads but exclusive writes.
+type SafeAggregateSet struct {
+	mu  sync.RWMutex
+	set *AggregateSet
+}
+
+// NewSafeAggregateSet creates a new thread-safe aggregate set.
+func NewSafeAggregateSet() *SafeAggregateSet {
+	return &SafeAggregateSet{
+		set: NewAggregateSet(),
+	}
+}
+
+// Aggregate data to the aggregate set with thread safety.
+func (s *SafeAggregateSet) Aggregate(key string, agg AggregateOperation, value string, clientAggregation bool) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.set.Aggregate(key, agg, value, clientAggregation)
+}
+
+// IncrementSamples increments the sample count safely.
+func (s *SafeAggregateSet) IncrementSamples() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.set.Samples++
+}
+
+// Clone creates a deep copy of the aggregate set.
+// This is useful for serialization to avoid holding locks for too long.
+func (s *SafeAggregateSet) Clone() *AggregateSet {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	clone := &AggregateSet{
+		Samples: s.set.Samples,
+		FValues: make(map[string]float64, len(s.set.FValues)),
+		SValues: make(map[string]string, len(s.set.SValues)),
+	}
+
+	// Deep copy the maps
+	for k, v := range s.set.FValues {
+		clone.FValues[k] = v
+	}
+	for k, v := range s.set.SValues {
+		clone.SValues[k] = v
+	}
+
+	return clone
+}
+
+// Serialize the aggregate set safely. Returns whether the message was sent
+// before the context was cancelled; mirrors AggregateSet.Serialize.
+func (s *SafeAggregateSet) Serialize(ctx context.Context, groupKey string, ch chan<- string) bool {
+	// Clone the set to avoid holding the lock during serialization
+	clone := s.Clone()
+	return clone.Serialize(ctx, groupKey, ch)
+}
+
+// GetSamples returns the current sample count safely.
+func (s *SafeAggregateSet) GetSamples() int {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	return s.set.Samples
+}
-- 
cgit v1.2.3