Maven/internal/store/conncap_test.go

package store

import (
	"context"
	"database/sql"
	"fmt"
	"path/filepath"
	"sort"
	"sync"
	"sync/atomic"
	"testing"
	"time"
)

// conncap_test.go measures whether a read queues behind a write at
// SetMaxOpenConns(1), which is what openAt sets (V-642). It is a measurement
// harness, not an assertion: the numbers it prints are the evidence, and the
// decision to move the cap or leave it belongs in docs/evals.
//
// Run it with -v, and note that it is skipped under -short because it spends
// seconds on purpose.

// openCapped opens a plaintext store at the given connection cap. In-package,
// so it can reach the handle openAt caps at 1.
func openCapped(t *testing.T, cap int) *Store {
	t.Helper()
	path := filepath.Join(t.TempDir(), "cap.db")
	db, err := openAt(context.Background(), path)
	if err != nil {
		t.Fatalf("openAt: %v", err)
	}
	db.SetMaxOpenConns(cap)
	s := &Store{db: db}
	t.Cleanup(func() { _ = s.Close() })
	return s
}

func percentile(d []time.Duration, p float64) time.Duration {
	if len(d) == 0 {
		return 0
	}
	i := int(float64(len(d)-1) * p)
	return d[i]
}

// seedFacts writes n facts so a read has rows to decode.
func seedFacts(t *testing.T, s *Store, n int) {
	t.Helper()
	ctx := context.Background()
	now := time.Now().UTC()
	for i := 0; i < n; i++ {
		key := fmt.Sprintf("seed_%d", i)
		if _, err := s.SetValue(ctx, KindSelf, key, "tap:test",
			map[string]int{"ml": i}, now.Add(time.Duration(i)*time.Millisecond)); err != nil {
			t.Fatalf("seed %d: %v", i, err)
		}
	}
}

// measureReadsUnderWrites reports read latency percentiles while a writer
// writes at a fixed pace. The pace matters: an unpaced writer completes a
// different number of writes at each cap, because at a higher cap it competes
// with the readers for the write lock instead of taking turns on one
// connection. Two runs that did different work cannot be compared.
// It runs for a fixed wall-clock window rather than a fixed read count, so the
// paced writer does the same work at every cap. Tying the window to a read
// count made the faster configuration receive fewer writes.
func measureReadsUnderWrites(t *testing.T, s *Store, window, pace time.Duration) []time.Duration {
	t.Helper()
	ctx := context.Background()

	var stop atomic.Bool
	var writes atomic.Int64
	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		defer wg.Done()
		now := time.Now().UTC()
		for i := 0; !stop.Load(); i++ {
			key := fmt.Sprintf("hot_%d", i%16)
			if _, err := s.SetValue(ctx, KindSelf, key, "tap:test",
				map[string]int{"n": i}, now.Add(time.Duration(i)*time.Millisecond)); err != nil {
				t.Errorf("write: %v", err)
				return
			}
			writes.Add(1)
			time.Sleep(pace)
		}
	}()

	var lat []time.Duration
	deadline := time.Now().Add(window)
	for time.Now().Before(deadline) {
		start := time.Now()
		if _, err := s.RecentFacts(ctx, 50); err != nil {
			t.Fatalf("RecentFacts: %v", err)
		}
		lat = append(lat, time.Since(start))
	}
	stop.Store(true)
	wg.Wait()
	t.Logf("in %v: %d reads, %d writes", window, len(lat), writes.Load())

	sort.Slice(lat, func(i, j int) bool { return lat[i] < lat[j] })
	return lat
}

// TestConnCap_ReadLatencyUnderWrites is the V-642 measurement: read latency at
// cap 1 against cap 4, same workload, same schema, same driver.
func TestConnCap_ReadLatencyUnderWrites(t *testing.T) {
	if testing.Short() {
		t.Skip("measurement harness; runs for seconds")
	}
	for _, cap := range []int{1, 4} {
		t.Run(fmt.Sprintf("cap=%d", cap), func(t *testing.T) {
			s := openCapped(t, cap)
			seedFacts(t, s, 500)
			lat := measureReadsUnderWrites(t, s, 2*time.Second, 2*time.Millisecond)
			t.Logf("cap=%d reads=%d p50=%v p95=%v max=%v",
				cap, len(lat), percentile(lat, 0.50), percentile(lat, 0.95), lat[len(lat)-1])
		})
	}
}

// TestConnCap_ReadBlocksBehindOpenSnapshot is the sharper claim: at cap 1 an
// open read-only transaction holds the only connection, so an unrelated read
// cannot proceed until it commits. This is why the store exposes no way to
// begin one — `Store.DB` used to, and was deleted in V-642 with no caller. The
// test stays as the reason, so re-adding that seam fails a measurement rather
// than shipping a stall.
func TestConnCap_ReadBlocksBehindOpenSnapshot(t *testing.T) {
	if testing.Short() {
		t.Skip("measurement harness; waits on a timeout")
	}
	for _, cap := range []int{1, 4} {
		t.Run(fmt.Sprintf("cap=%d", cap), func(t *testing.T) {
			s := openCapped(t, cap)
			seedFacts(t, s, 50)

			tx, err := s.db.BeginTx(context.Background(), &sql.TxOptions{ReadOnly: true})
			if err != nil {
				t.Fatalf("BeginTx: %v", err)
			}
			defer func() { _ = tx.Rollback() }()

			ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
			defer cancel()
			start := time.Now()
			_, err = s.RecentFacts(ctx, 10)
			t.Logf("cap=%d read alongside an open snapshot: waited %v, err=%v",
				cap, time.Since(start).Round(time.Millisecond), err)
		})
	}
}