test: DTail fork — integration test suite and fixtures

Squashed development of the integration test suite (integrationtests/) covering
DCat, DGrep, DMap (serverless + server mode), DTail follow, DServer, DTailHealth,
journal source reads, auth-key fast reconnect, interactive query reload,
client-deadline/timeout behaviour, and the single-mode read/output path.

Includes real test fixtures (dserver*.cfg, dmap_csv_multifile_*.csv.in,
test_server_*.json, *.expected golden files) and deterministic synchronization
helpers (waitContains-style barriers) replacing racy fixed-timing assertions.

Accidental debug/output dumps that earlier commits added here (captured client
output, strace logs, ad-hoc turbo_test_output/manual_output/test_output files,
throwaway debug scripts) are intentionally excluded and gitignored.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

1 files changed, 535 insertions, 0 deletions

diff --git a/integrationtests/interactive_runtime_query_test.go b/integrationtests/interactive_runtime_query_test.go
new file mode 100644
index 0000000..ca020b1
--- /dev/null
+++ b/integrationtests/interactive_runtime_query_test.go
@@ -0,0 +1,535 @@
+package integrationtests
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+	"os/exec"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+)
+
+type interactiveStep struct {
+	// At is a fixed delay (from the writer start) before Input is sent. It is
+	// only consulted when WaitFor is empty.
+	At time.Duration
+	// WaitFor, when set, makes the step block until this substring appears in
+	// the accumulated PTY output before Input is sent (bounded by a timeout).
+	// This replaces racy fixed-offset timing with deterministic
+	// synchronization on observed output.
+	WaitFor string
+	Input   string
+}
+
+type processOutputCollector struct {
+	mu    sync.Mutex
+	lines []string
+}
+
+func TestDTailInteractiveReloadReusesSessionAndDropsLateOldMatches(t *testing.T) {
+	skipIfNotIntegrationTest(t)
+
+	testLogger := NewTestLogger("TestDTailInteractiveReloadReusesSessionAndDropsLateOldMatches")
+	defer testLogger.WriteLogFile()
+	cleanupTmpFiles(t)
+
+	ctx, cancel := createTestContextWithTimeout(t)
+	ctx = WithTestLogger(ctx, testLogger)
+	defer cancel()
+
+	followFile := "interactive_dtail_reload.tmp"
+	if err := os.WriteFile(followFile, nil, 0600); err != nil {
+		t.Fatalf("unable to create follow file: %v", err)
+	}
+	cleanupFiles(t, followFile, "interactive_dtail_reload.stdout.tmp")
+
+	port := getUniquePortNumber()
+	serverStdout, serverStderr, _, err := startCommand(ctx, t, "", "../dserver",
+		"--cfg", "none",
+		"--logger", "stdout",
+		"--logLevel", "debug",
+		"--bindAddress", "localhost",
+		"--port", fmt.Sprintf("%d", port),
+	)
+	if err != nil {
+		t.Fatalf("start dserver: %v", err)
+	}
+	serverLogs := startProcessOutputCollector(ctx, serverStdout, serverStderr)
+	if err := waitForServerReady(ctx, "localhost", port); err != nil {
+		t.Fatalf("wait for dserver: %v", err)
+	}
+	serverLogs.reset()
+
+	writerDone := make(chan error, 1)
+	go func() {
+		if err := waitForCollectorSubstring(ctx, serverLogs, "Start reading|"+followFile+"|"+followFile); err != nil {
+			writerDone <- err
+			return
+		}
+		writerDone <- appendLinesOnSchedule(ctx, followFile, []interactiveStep{
+			{At: 100 * time.Millisecond, Input: "ERROR initial"},
+			{At: 3000 * time.Millisecond, Input: "ERROR late"},
+			{At: 3200 * time.Millisecond, Input: "WARN live"},
+		})
+	}()
+
+	clientOutput, err := runInteractivePTYCommand(ctx, []string{
+		"../dtail",
+		"--cfg", "none",
+		"--logger", "stdout",
+		"--logLevel", "info",
+		"--servers", fmt.Sprintf("localhost:%d", port),
+		"--files", followFile,
+		"--grep", "ERROR",
+		"--plain",
+		"--trustAllHosts",
+		"--interactive-query",
+	}, []interactiveStep{
+		{At: 3 * time.Second, Input: ":reload --grep WARN"},
+		{At: 6 * time.Second, Input: ":quit"},
+	})
+	if err != nil {
+		t.Fatalf("run interactive dtail: %v\noutput:\n%s", err, clientOutput)
+	}
+
+	if err := <-writerDone; err != nil {
+		t.Fatalf("write follow file: %v", err)
+	}
+
+	if !strings.Contains(clientOutput, "ERROR initial") {
+		t.Fatalf("expected initial ERROR line in output:\n%s", clientOutput)
+	}
+	if !strings.Contains(clientOutput, "WARN live") {
+		t.Fatalf("expected WARN line after reload in output:\n%s", clientOutput)
+	}
+	if strings.Contains(clientOutput, "ERROR late") {
+		t.Fatalf("unexpected stale ERROR line after reload:\n%s", clientOutput)
+	}
+	if !strings.Contains(clientOutput, "reload applied successfully") {
+		t.Fatalf("expected reload success message in output:\n%s", clientOutput)
+	}
+	if countSubstring(serverLogs.snapshot(), "Creating new server handler") != 1 {
+		t.Fatalf("expected one SSH session on the server, logs:\n%s", strings.Join(serverLogs.snapshot(), "\n"))
+	}
+}
+
+func TestDTailInteractiveReloadReusesSessionOnImmediateBoundaryAndDropsLateOldMatches(t *testing.T) {
+	skipIfNotIntegrationTest(t)
+
+	testLogger := NewTestLogger("TestDTailInteractiveReloadReusesSessionOnImmediateBoundaryAndDropsLateOldMatches")
+	defer testLogger.WriteLogFile()
+	cleanupTmpFiles(t)
+
+	ctx, cancel := createTestContextWithTimeout(t)
+	ctx = WithTestLogger(ctx, testLogger)
+	defer cancel()
+
+	followFile := "interactive_dtail_reload_immediate.tmp"
+	// Start with an empty follow file; the server tails appended lines (it does
+	// not replay pre-existing content), so all matches are fed as appends below.
+	if err := os.WriteFile(followFile, nil, 0600); err != nil {
+		t.Fatalf("unable to create follow file: %v", err)
+	}
+	cleanupFiles(t, followFile, "interactive_dtail_reload_immediate.stdout.tmp")
+
+	port := getUniquePortNumber()
+	serverStdout, serverStderr, _, err := startCommand(ctx, t, "", "../dserver",
+		"--cfg", "none",
+		"--logger", "stdout",
+		"--logLevel", "debug",
+		"--bindAddress", "localhost",
+		"--port", fmt.Sprintf("%d", port),
+	)
+	if err != nil {
+		t.Fatalf("start dserver: %v", err)
+	}
+	serverLogs := startProcessOutputCollector(ctx, serverStdout, serverStderr)
+	if err := waitForServerReady(ctx, "localhost", port); err != nil {
+		t.Fatalf("wait for dserver: %v", err)
+	}
+	serverLogs.reset()
+
+	// Feed follow-file content at deterministic points instead of on fixed
+	// offsets from an independent clock (the previous racy design, which under
+	// load could append the boundary line after the reload had already switched
+	// the grep to WARN, so it never matched -> "expected first-generation ERROR
+	// line before reload").
+	//
+	// Sequence, each step gated on an observed barrier:
+	//  1. wait for the first "Start reading" (session tailing), then append the
+	//     first-generation "ERROR boundary" match. The reload step blocks until
+	//     this line is seen in the client output, so it always precedes reload.
+	//  2. wait for the second "Start reading" (the reload dispatches a fresh
+	//     read command under the new grep-WARN generation; that log is emitted
+	//     as the new read begins, a sound barrier), then append the post-reload
+	//     lines. "ERROR late" must be dropped (no longer matches WARN) and
+	//     "WARN live" must appear.
+	readNeedle := "Start reading|" + followFile + "|" + followFile
+	writerDone := make(chan error, 1)
+	go func() {
+		if err := waitForCollectorSubstring(ctx, serverLogs, readNeedle); err != nil {
+			writerDone <- err
+			return
+		}
+		if err := appendLines(followFile, "ERROR boundary"); err != nil {
+			writerDone <- err
+			return
+		}
+		if err := waitForCollectorCount(ctx, serverLogs, readNeedle, 2); err != nil {
+			writerDone <- err
+			return
+		}
+		writerDone <- appendLines(followFile, "ERROR late", "WARN live")
+	}()
+
+	clientOutput, err := runInteractivePTYCommand(ctx, []string{
+		"../dtail",
+		"--cfg", "none",
+		"--logger", "stdout",
+		"--logLevel", "info",
+		"--servers", fmt.Sprintf("localhost:%d", port),
+		"--files", followFile,
+		"--grep", "ERROR",
+		"--plain",
+		"--trustAllHosts",
+		"--interactive-query",
+	}, []interactiveStep{
+		// Reload only after the first-generation ERROR match has actually been
+		// emitted, so the boundary line deterministically precedes the reload
+		// regardless of connection-setup latency under load.
+		{WaitFor: "ERROR boundary", Input: ":reload --grep WARN"},
+		// Quit only after the new-generation WARN match has been emitted, so the
+		// captured output deterministically contains it before the session ends.
+		{WaitFor: "WARN live", Input: ":quit"},
+	})
+	if err != nil {
+		t.Fatalf("run interactive dtail: %v\noutput:\n%s", err, clientOutput)
+	}
+
+	if err := <-writerDone; err != nil {
+		t.Fatalf("write follow file: %v", err)
+	}
+
+	if !strings.Contains(clientOutput, "WARN live") {
+		t.Fatalf("expected WARN line after reload in output:\n%s", clientOutput)
+	}
+	if !strings.Contains(clientOutput, "ERROR boundary") {
+		t.Fatalf("expected first-generation ERROR line before reload in output:\n%s", clientOutput)
+	}
+	if strings.Contains(clientOutput, "ERROR late") {
+		t.Fatalf("unexpected stale ERROR line after reload:\n%s", clientOutput)
+	}
+	if !strings.Contains(clientOutput, "reload applied successfully") {
+		t.Fatalf("expected reload success message in output:\n%s", clientOutput)
+	}
+	if boundaryIdx := strings.Index(clientOutput, "ERROR boundary"); boundaryIdx == -1 || boundaryIdx > strings.Index(clientOutput, "reload applied successfully") {
+		t.Fatalf("expected first-generation ERROR output to precede reload success:\n%s", clientOutput)
+	}
+	if countSubstring(serverLogs.snapshot(), "Creating new server handler") != 1 {
+		t.Fatalf("expected one SSH session on the server, logs:\n%s", strings.Join(serverLogs.snapshot(), "\n"))
+	}
+}
+
+func TestDGrepInteractiveReloadReusesSessionAfterCompletedRead(t *testing.T) {
+	skipIfNotIntegrationTest(t)
+
+	testLogger := NewTestLogger("TestDGrepInteractiveReloadReusesSessionAfterCompletedRead")
+	defer testLogger.WriteLogFile()
+	cleanupTmpFiles(t)
+
+	ctx, cancel := createTestContextWithTimeout(t)
+	ctx = WithTestLogger(ctx, testLogger)
+	defer cancel()
+
+	inputFile := "interactive_dgrep_reload.tmp"
+	if err := os.WriteFile(inputFile, []byte("ERROR first\nWARN second\n"), 0600); err != nil {
+		t.Fatalf("unable to create input file: %v", err)
+	}
+	cleanupFiles(t, inputFile, "interactive_dgrep_reload.stdout.tmp")
+
+	port := getUniquePortNumber()
+	serverStdout, serverStderr, _, err := startCommand(ctx, t, "", "../dserver",
+		"--cfg", "none",
+		"--logger", "stdout",
+		"--logLevel", "debug",
+		"--bindAddress", "localhost",
+		"--port", fmt.Sprintf("%d", port),
+	)
+	if err != nil {
+		t.Fatalf("start dserver: %v", err)
+	}
+	serverLogs := startProcessOutputCollector(ctx, serverStdout, serverStderr)
+	if err := waitForServerReady(ctx, "localhost", port); err != nil {
+		t.Fatalf("wait for dserver: %v", err)
+	}
+	serverLogs.reset()
+
+	clientOutput, err := runInteractivePTYCommand(ctx, []string{
+		"../dgrep",
+		"--cfg", "none",
+		"--logger", "stdout",
+		"--logLevel", "info",
+		"--servers", fmt.Sprintf("localhost:%d", port),
+		"--files", inputFile,
+		"--grep", "ERROR",
+		"--plain",
+		"--trustAllHosts",
+		"--interactive-query",
+	}, []interactiveStep{
+		{At: 3500 * time.Millisecond, Input: ":reload --grep WARN"},
+		{At: 5500 * time.Millisecond, Input: ":quit"},
+	})
+	if err != nil {
+		t.Fatalf("run interactive dgrep: %v\noutput:\n%s", err, clientOutput)
+	}
+
+	if !strings.Contains(clientOutput, "ERROR first") {
+		t.Fatalf("expected initial grep result in output:\n%s", clientOutput)
+	}
+	if !strings.Contains(clientOutput, "WARN second") {
+		t.Fatalf("expected reloaded grep result in output:\n%s", clientOutput)
+	}
+	if !strings.Contains(clientOutput, "reload applied successfully") {
+		t.Fatalf("expected reload success message in output:\n%s", clientOutput)
+	}
+	if countSubstring(serverLogs.snapshot(), "Creating new server handler") != 1 {
+		t.Fatalf("expected one SSH session on the server, logs:\n%s", strings.Join(serverLogs.snapshot(), "\n"))
+	}
+}
+
+func startProcessOutputCollector(ctx context.Context, stdoutCh, stderrCh <-chan string) *processOutputCollector {
+	collector := &processOutputCollector{}
+	collect := func(ch <-chan string) {
+		for {
+			select {
+			case line, ok := <-ch:
+				if !ok {
+					return
+				}
+				collector.append(line)
+			case <-ctx.Done():
+				return
+			}
+		}
+	}
+	go collect(stdoutCh)
+	go collect(stderrCh)
+	return collector
+}
+
+func (c *processOutputCollector) append(line string) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.lines = append(c.lines, line)
+}
+
+func (c *processOutputCollector) snapshot() []string {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	out := make([]string, len(c.lines))
+	copy(out, c.lines)
+	return out
+}
+
+func (c *processOutputCollector) reset() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.lines = c.lines[:0]
+}
+
+func appendLinesOnSchedule(ctx context.Context, path string, steps []interactiveStep) error {
+	fd, err := os.OpenFile(path, os.O_WRONLY|os.O_APPEND, 0600)
+	if err != nil {
+		return err
+	}
+	defer fd.Close()
+
+	start := time.Now()
+	for _, step := range steps {
+		wait := time.Until(start.Add(step.At))
+		if wait > 0 {
+			timer := time.NewTimer(wait)
+			select {
+			case <-ctx.Done():
+				timer.Stop()
+				return ctx.Err()
+			case <-timer.C:
+			}
+		}
+
+		if _, err := fd.WriteString(step.Input + "\n"); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func runInteractivePTYCommand(ctx context.Context, argv []string, steps []interactiveStep) (string, error) {
+	if _, err := exec.LookPath("python3"); err != nil {
+		return "", fmt.Errorf("python3 is required for PTY integration tests: %w", err)
+	}
+
+	script := `
+import json
+import os
+import pty
+import sys
+import threading
+import time
+
+argv = json.loads(os.environ["DTAIL_PTY_ARGV"])
+steps = json.loads(os.environ["DTAIL_PTY_STEPS"])
+
+pid, fd = pty.fork()
+if pid == 0:
+    os.execv(argv[0], argv)
+
+# Shared output buffer, populated by the reader loop below and consulted by the
+# writer thread so a step can wait for a marker to appear before sending input.
+# Declared before the writer starts so a wait_for step never races the buffer.
+output = bytearray()
+
+# Upper bound for how long a wait_for step blocks. The Go-side test context also
+# bounds the whole process, so a genuine product hang still fails rather than
+# hanging forever; on timeout the input is sent anyway so the pending Go
+# assertion reports the real problem instead of the harness deadlocking.
+WAIT_FOR_TIMEOUT = 30.0
+
+def writer():
+    start = time.monotonic()
+    for step in steps:
+        wait_for = step.get("wait_for") or ""
+        if wait_for:
+            needle = wait_for.encode("utf-8")
+            deadline = time.monotonic() + WAIT_FOR_TIMEOUT
+            while needle not in bytes(output):
+                if time.monotonic() >= deadline:
+                    break
+                time.sleep(0.01)
+        else:
+            wait = (step["at_ms"] / 1000.0) - (time.monotonic() - start)
+            if wait > 0:
+                time.sleep(wait)
+        data = step["input"]
+        if not data.endswith("\n"):
+            data += "\n"
+        os.write(fd, data.encode("utf-8"))
+
+threading.Thread(target=writer, daemon=True).start()
+
+while True:
+    try:
+        chunk = os.read(fd, 4096)
+    except OSError:
+        break
+    if not chunk:
+        break
+    output.extend(chunk)
+
+_, status = os.waitpid(pid, 0)
+sys.stdout.buffer.write(output)
+if os.WIFEXITED(status):
+    sys.exit(os.WEXITSTATUS(status))
+if os.WIFSIGNALED(status):
+    sys.exit(128 + os.WTERMSIG(status))
+sys.exit(1)
+`
+
+	encodedSteps := make([]map[string]any, 0, len(steps))
+	for _, step := range steps {
+		encodedSteps = append(encodedSteps, map[string]any{
+			"at_ms":    step.At.Milliseconds(),
+			"wait_for": step.WaitFor,
+			"input":    step.Input,
+		})
+	}
+
+	argvPayload, err := json.Marshal(argv)
+	if err != nil {
+		return "", err
+	}
+	stepsPayload, err := json.Marshal(encodedSteps)
+	if err != nil {
+		return "", err
+	}
+
+	cmd := exec.CommandContext(ctx, "python3", "-c", script)
+	cmd.Env = append(os.Environ(),
+		"DTAIL_PTY_ARGV="+string(argvPayload),
+		"DTAIL_PTY_STEPS="+string(stepsPayload),
+	)
+	output, err := cmd.CombinedOutput()
+	return string(output), err
+}
+
+func countSubstring(lines []string, needle string) int {
+	count := 0
+	for _, line := range lines {
+		if strings.Contains(line, needle) {
+			count++
+		}
+	}
+	return count
+}
+
+// waitForCollectorCount blocks until at least min lines containing needle have
+// been collected, or the context is done. It is used to wait for the Nth
+// occurrence of a server log line (e.g. the second "Start reading" emitted when
+// a reload dispatches a fresh read command under the new generation).
+func waitForCollectorCount(ctx context.Context, collector *processOutputCollector, needle string, min int) error {
+	ticker := time.NewTicker(10 * time.Millisecond)
+	defer ticker.Stop()
+
+	for {
+		if countSubstring(collector.snapshot(), needle) >= min {
+			return nil
+		}
+
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-ticker.C:
+		}
+	}
+}
+
+// appendLines appends the given lines (each terminated with a newline) to the
+// file at path. Used to feed follow-file content at deterministic points once a
+// synchronization barrier has been observed, rather than on a fixed schedule.
+func appendLines(path string, lines ...string) error {
+	fd, err := os.OpenFile(path, os.O_WRONLY|os.O_APPEND, 0600)
+	if err != nil {
+		return err
+	}
+	defer fd.Close()
+
+	for _, line := range lines {
+		if _, err := fd.WriteString(line + "\n"); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func waitForCollectorSubstring(ctx context.Context, collector *processOutputCollector, needle string) error {
+	ticker := time.NewTicker(10 * time.Millisecond)
+	defer ticker.Stop()
+
+	for {
+		for _, line := range collector.snapshot() {
+			if strings.Contains(line, needle) {
+				return nil
+			}
+		}
+
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-ticker.C:
+		}
+	}
+}