ci: exercise lifecycle-sensitive paths

.github/workflows/ci.yml

@@ -3,12 +3,15 @@ name: ci
 on:
   pull_request:
     branches: ["main"]
+  schedule:
+    - cron: "17 3 * * 1"
 
 env:
   GO_VERSION: "1.27.0"
 
 jobs:
   check:
+    if: ${{ github.event_name == 'pull_request' }}
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v7
@@ -37,6 +40,7 @@ jobs:
           install-go: false
 
   test:
+    if: ${{ github.event_name == 'pull_request' }}
     runs-on: ${{ matrix.os }}
     strategy:
       matrix:
@@ -54,7 +58,23 @@ jobs:
       - name: Test
         run: go test -v -parallel 8 ./...
 
+  race:
+    if: ${{ github.event_name == 'pull_request' }}
+    name: race (linux)
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v7
+
+      - name: Setup Go
+        uses: actions/setup-go@v7
+        with:
+          go-version: ${{ env.GO_VERSION }}
+
+      - name: Race test lifecycle-sensitive packages
+        run: go test -race -count=1 -parallel 8 ./internal/client ./internal/fetch ./internal/grpc ./internal/pager ./internal/resolver ./internal/ws
+
   test-windows:
+    if: ${{ github.event_name == 'pull_request' }}
     name: test (windows-latest)
     needs: build-windows
     runs-on: windows-latest
@@ -80,6 +100,7 @@ jobs:
         run: go test -v -parallel 8 ./...
 
   build-windows:
+    if: ${{ github.event_name == 'pull_request' }}
     name: build (windows-amd64)
     runs-on: ubuntu-latest
     steps:
@@ -108,6 +129,7 @@ jobs:
           retention-days: 1
 
   build:
+    if: ${{ github.event_name == 'pull_request' }}
     runs-on: ubuntu-latest
     strategy:
       matrix:
@@ -147,3 +169,30 @@ jobs:
           GOARCH: ${{ matrix.goarch }}
           CGO_ENABLED: "0"
         run: go install -trimpath -ldflags="-s -w"
+
+  scheduled-smoke:
+    if: ${{ github.event_name == 'schedule' }}
+    name: scheduled fuzz and benchmark smoke
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v7
+
+      - name: Setup Go
+        uses: actions/setup-go@v7
+        with:
+          go-version: ${{ env.GO_VERSION }}
+
+      - name: Fuzz DNS wire decoder
+        run: go test -run '^$' -fuzz FuzzDecodeDNSWire -fuzztime 5s ./internal/resolver
+
+      - name: Fuzz gRPC framing
+        run: go test -run '^$' -fuzz FuzzUnframe -fuzztime 5s ./internal/grpc
+
+      - name: Fuzz WebSocket piped-line framing
+        run: go test -run '^$' -fuzz FuzzReadBoundedLine -fuzztime 5s ./internal/ws
+
+      - name: Benchmark terminal streaming
+        run: go test -run '^$' -bench '^BenchmarkTerminalSafeReader' -benchtime 1x ./internal/fetch
+
+      - name: Benchmark gRPC framing
+        run: go test -run '^$' -bench '^BenchmarkFrame' -benchtime 1x ./internal/grpc

internal/grpc/framing_fuzz_test.go

@@ -0,0 +1,18 @@
+package grpc
+
+import "testing"
+
+func FuzzUnframe(f *testing.F) {
+	f.Add(Frame([]byte("hello"), false))
+	f.Add(Frame([]byte("compressed"), true))
+	f.Add([]byte{0x02, 0, 0, 0, 0})
+	f.Add([]byte{0, 0x10, 0, 0, 0})
+
+	f.Fuzz(func(t *testing.T, frame []byte) {
+		const maxFuzzInput = 1 << 20
+		if len(frame) > maxFuzzInput {
+			frame = frame[:maxFuzzInput]
+		}
+		_, _, _ = Unframe(frame)
+	})
+}

internal/grpc/framing_test.go

@@ -3,6 +3,7 @@ package grpc
 import (
 	"bytes"
 	"compress/gzip"
+	"fmt"
 	"io"
 	"testing"
 )
@@ -335,3 +336,23 @@ func TestReadFrameRoundTrip(t *testing.T) {
 		}
 	}
 }
+
+var benchmarkFrameResult []byte
+
+func BenchmarkFrame(b *testing.B) {
+	for _, size := range []int{0, 1 << 10, 64 << 10, 1 << 20} {
+		b.Run(fmt.Sprintf("%dB", size), func(b *testing.B) {
+			data := bytes.Repeat([]byte{'x'}, size)
+			b.SetBytes(int64(len(data)))
+			b.ReportAllocs()
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				var err error
+				benchmarkFrameResult, err = FrameChecked(data, false)
+				if err != nil {
+					b.Fatal(err)
+				}
+			}
+		})
+	}
+}

internal/pager/pager.go

@@ -77,16 +77,31 @@ func StreamContext(ctx context.Context, src io.Reader, mode core.PagerMode, stdo
 	// An interactive pager must remain in the terminal's foreground process
 	// group. Otherwise it cannot receive keystrokes such as `q`; the parent
 	// still owns the pager's data pipe, so the command appears to hang.
-	configureProcess(cmd, stdoutTerminal)
+	if err := configureProcess(cmd, stdoutTerminal); err != nil {
+		return fmt.Errorf("unable to configure pager: %w", err)
+	}
 	cmd.Stdout = dst
 	cmd.Stderr = os.Stderr
 	stdin, err := cmd.StdinPipe()
 	if err != nil {
+		releaseProcess(cmd)
 		return err
 	}
 	if err := cmd.Start(); err != nil {
+		terminateProcessTree(cmd)
+		if cmd.Process != nil {
+			_ = cmd.Wait()
+		}
+		releaseProcess(cmd)
 		return fmt.Errorf("unable to start pager: %w", err)
 	}
+	if err := attachProcess(cmd); err != nil {
+		terminateProcessTree(cmd)
+		_ = cmd.Wait()
+		releaseProcess(cmd)
+		return fmt.Errorf("unable to attach pager: %w", err)
+	}
+	defer releaseProcess(cmd)
 
 	// A non-closable reader cannot be interrupted by this package. Keep its
 	// copy synchronous so cancellation cannot strand a producer goroutine.

internal/pager/pager_test.go

@@ -3,7 +3,10 @@ package pager
 import (
 	"context"
 	"io"
+	"os"
+	"path/filepath"
 	"runtime"
+	"strconv"
 	"strings"
 	"testing"
 	"time"
@@ -75,23 +78,83 @@ func TestStreamContextTerminatesPagerProcessGroup(t *testing.T) {
 	if testing.Short() || runtime.GOOS == "windows" {
 		t.Skip("starts a Unix subprocess")
 	}
-	t.Setenv("PAGER", "sh -c 'sleep 30'")
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
+
+	fixtureDir := t.TempDir()
+	readyPath := filepath.Join(fixtureDir, "ready")
+	childPIDPath := filepath.Join(fixtureDir, "child.pid")
+	pagerPath := filepath.Join(fixtureDir, "pager.sh")
+	if err := os.WriteFile(pagerPath, []byte("#!/bin/sh\nsleep 30 &\nchild_pid=$!\nprintf '%s\\n' \"$child_pid\" > \"$FETCH_TEST_PAGER_CHILD_PID\"\nprintf ready > \"$FETCH_TEST_PAGER_READY\"\nwait \"$child_pid\"\n"), 0o700); err != nil {
+		t.Fatal(err)
+	}
+	t.Setenv("FETCH_TEST_PAGER_READY", readyPath)
+	t.Setenv("FETCH_TEST_PAGER_CHILD_PID", childPIDPath)
+	t.Setenv("PAGER", pagerPath)
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
 	done := make(chan error, 1)
 	go func() {
 		done <- StreamContext(ctx, strings.NewReader("help"), core.PagerOn, false, false, io.Discard)
 	}()
-	time.Sleep(50 * time.Millisecond)
+	childPID := 0
+	streamDone := false
+	t.Cleanup(func() {
+		cancel()
+		if !streamDone {
+			select {
+			case <-done:
+			case <-time.After(time.Second):
+				t.Log("pager cleanup exceeded its deadline")
+			}
+		}
+		if childPID != 0 && !testProcessExited(childPID) {
+			killTestProcess(childPID)
+			if !waitForTestProcessExit(childPID, time.Second) {
+				t.Logf("pager child process %d did not exit during cleanup", childPID)
+			}
+		}
+	})
+
+	readyDeadline := time.NewTimer(time.Second)
+	defer readyDeadline.Stop()
+	readyTicker := time.NewTicker(5 * time.Millisecond)
+	defer readyTicker.Stop()
+	for {
+		if _, err := os.Stat(readyPath); err == nil {
+			break
+		}
+		select {
+		case <-readyDeadline.C:
+			t.Fatal("pager fixture did not start before the deadline")
+		case <-ctx.Done():
+			t.Fatal("pager fixture did not start before the deadline")
+		case <-readyTicker.C:
+		}
+	}
+
+	pidData, err := os.ReadFile(childPIDPath)
+	if err != nil {
+		t.Fatalf("read pager child PID: %v", err)
+	}
+	childPID, err = strconv.Atoi(strings.TrimSpace(string(pidData)))
+	if err != nil || childPID <= 0 {
+		t.Fatalf("invalid pager child PID %q", pidData)
+	}
+	if testProcessExited(childPID) {
+		t.Fatalf("pager child process %d exited before cancellation", childPID)
+	}
+
 	cancel()
 	select {
-	case err := <-done:
-		if err == nil {
+	case streamErr := <-done:
+		streamDone = true
+		if streamErr == nil {
 			t.Fatal("canceled pager returned nil")
 		}
 	case <-time.After(2 * time.Second):
 		t.Fatal("pager process group did not terminate")
 	}
+	if !waitForTestProcessExit(childPID, time.Second) {
+		t.Fatalf("pager child process %d survived cancellation", childPID)
+	}
 }
 
 func TestStreamContextReportsPagerStartFailure(t *testing.T) {