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

diff --git a/internal/mapr/aggregateset.go b/internal/mapr/aggregateset.go
index c50c7a1..3281353 100644
--- a/internal/mapr/aggregateset.go
+++ b/internal/mapr/aggregateset.go
@@ -46,6 +46,10 @@ func (s *AggregateSet) Merge(query *Query, set *AggregateSet) error {
 		case Sum:
 			fallthrough
 		case Avg:
+			fallthrough
+		case Percentage:
+			fallthrough
+		case Percentile:
 			value := set.FValues[storage]
 			s.addFloat(storage, value)
 		case Min:
@@ -67,21 +71,25 @@ func (s *AggregateSet) Merge(query *Query, set *AggregateSet) error {
 	return nil
 }
 
-// Serialize the aggregate set so it can be sent over the wire.
-func (s *AggregateSet) Serialize(ctx context.Context, groupKey string, ch chan<- string) {
+// Serialize the aggregate set so it can be sent over the wire. Returns true
+// when the serialized message was successfully sent, and false when the
+// context was cancelled before the send completed. Callers that own the
+// source state (e.g. Aggregate) must re-merge unsent sets so data is
+// not silently lost.
+func (s *AggregateSet) Serialize(ctx context.Context, groupKey string, ch chan<- string) bool {
 	dlog.Common.Trace("Serialising mapr.AggregateSet", s)
 	sb := pool.BuilderBuffer.Get().(*strings.Builder)
 	defer pool.RecycleBuilderBuffer(sb)
 
 	sb.WriteString(groupKey)
 	sb.WriteString(protocol.AggregateDelimiter)
-	sb.WriteString(fmt.Sprintf("%d", s.Samples))
+	sb.WriteString(strconv.Itoa(s.Samples))
 	sb.WriteString(protocol.AggregateDelimiter)
 
 	for k, v := range s.FValues {
 		sb.WriteString(k)
 		sb.WriteString(protocol.AggregateKVDelimiter)
-		sb.WriteString(fmt.Sprintf("%v", v))
+		sb.WriteString(strconv.FormatFloat(v, 'f', -1, 64))
 		sb.WriteString(protocol.AggregateDelimiter)
 	}
 
@@ -94,7 +102,9 @@ func (s *AggregateSet) Serialize(ctx context.Context, groupKey string, ch chan<-
 
 	select {
 	case ch <- sb.String():
+		return true
 	case <-ctx.Done():
+		return false
 	}
 }
 
@@ -177,6 +187,10 @@ func (s *AggregateSet) Aggregate(key string, agg AggregateOperation, value strin
 	case Sum:
 		fallthrough
 	case Avg:
+		fallthrough
+	case Percentage:
+		fallthrough
+	case Percentile:
 		s.addFloat(key, f)
 	case Min:
 		s.addFloatMin(key, f)