addrkit/test/e2e/e2e_test.go

//go:build e2e

// Package e2e contains process-level tests against a real registry snapshot.
//
// They build the CLI and the daemon, seed a temporary SQLite index from an
// EДДС Bulk NDJSON dump, start the HTTP service and exercise it with the whole
// registry (selfcheck) plus a large concurrent sample over HTTP.
//
// The data is production-derived and contains personal data. The tests never
// copy it into the repository: the input path is supplied through the
// ADDRKIT_E2E_REGISTRY environment variable, and every temporary file lives in
// t.TempDir().
//
// Run with:
//
//	ADDRKIT_E2E_REGISTRY=/path/to/address_registry.ndjson \
//	ADDRKIT_E2E_LOG=/path/to/edds_address_log.ndjson \
//	  go test -tags e2e -v ./test/e2e
package e2e

import (
	"bytes"
	"database/sql"
	"encoding/json"
	"fmt"
	"io"
	"math/rand"
	"net"
	"net/http"
	"os"
	"os/exec"
	"path/filepath"
	"runtime"
	"sort"
	"strings"
	"sync"
	"syscall"
	"testing"
	"time"

	"git.totmin.ru/en2zmax/addrkit/internal/adapters/registrystore"
	"git.totmin.ru/en2zmax/addrkit/internal/domain/address"
	"git.totmin.ru/en2zmax/addrkit/internal/domain/match"
)

const (
	defaultCity = "Новосибирск"

	// The service must answer canonical registry rows quickly. The limit is
	// deliberately generous: it catches pathological regressions, not jitter
	// on a shared CI machine.
	maxP99 = 100 * time.Millisecond
	maxRSS = 1 << 30 // 1 GiB
)

type serviceConfig struct {
	Addr        string        `json:"addr"`
	DataPath    string        `json:"data_path"`
	LogLevel    string        `json:"log_level"`
	Cities      []cityConfig  `json:"cities"`
	DefaultCity string        `json:"default_city"`
	Resolve     resolveConfig `json:"resolve"`
	LLM         llmConfig     `json:"llm"`
}

type cityConfig struct {
	Name    string   `json:"name"`
	Aliases []string `json:"aliases,omitempty"`
}

type resolveConfig struct {
	FuzzyThreshold  float64 `json:"fuzzy_threshold"`
	AmbiguityMargin float64 `json:"ambiguity_margin"`
	MaxCandidates   int     `json:"max_candidates"`
}

type llmConfig struct {
	Enabled bool `json:"enabled"`
}

type seedOutput struct {
	Entries int `json:"entries"`
	Stats   struct {
		Total           int            `json:"total"`
		Parsed          int            `json:"parsed"`
		Unparsed        int            `json:"unparsed"`
		Duplicates      int            `json:"duplicates"`
		DistinctStreets int            `json:"distinct_streets"`
		Cities          map[string]int `json:"cities"`
		UnparsedSample  []string       `json:"unparsed_sample"`
	} `json:"stats"`
}

type selfcheckReport struct {
	Entries   int            `json:"entries"`
	Exact     int            `json:"exact"`
	ExactRate float64        `json:"exact_rate"`
	NonExact  map[string]int `json:"non_exact"`
	Failures  []string       `json:"failures"`
}

type env struct {
	root       string
	binDir     string
	addrkit    string
	addrkitd   string
	db         string
	addr       string
	city       string
	parser     *address.Parser
	httpClient *http.Client
	seedReport *seedOutput
}

func TestProdNovosibirsk(t *testing.T) {
	registry := os.Getenv("ADDRKIT_E2E_REGISTRY")
	if registry == "" {
		t.Skip("ADDRKIT_E2E_REGISTRY is not set")
	}
	if _, err := os.Stat(registry); err != nil {
		t.Fatalf("registry dump: %v", err)
	}

	e := newEnv(t)
	entries := e.seed(t, registry)
	if len(entries) < 70_000 {
		t.Fatalf("seeded only %d entries; expected a full registry", len(entries))
	}
	if got := e.seedReport.Stats.Cities[e.city]; got != len(entries) {
		t.Fatalf("city %q covers %d of %d entries", e.city, got, len(entries))
	}
	t.Logf("seeded: total=%d parsed=%d unparsed=%d duplicates=%d streets=%d entries=%d",
		e.seedReport.Stats.Total, e.seedReport.Stats.Parsed, e.seedReport.Stats.Unparsed,
		e.seedReport.Stats.Duplicates, e.seedReport.Stats.DistinctStreets, len(entries))
	if e.seedReport.Stats.Unparsed*1000 > e.seedReport.Stats.Total {
		t.Errorf("unparsed rows = %d of %d, want <= 0.1%%", e.seedReport.Stats.Unparsed, e.seedReport.Stats.Total)
	}

	// Source rows without a house are indexed for street lookups but cannot
	// round-trip to "found"; the known EДДС formats ("27с1", "5 киоск") are
	// counted by selfcheck and excluded from the exact round-trip sample.
	resolvable := make([]address.Entry, 0, len(entries))
	for _, en := range entries {
		if en.House != "" {
			resolvable = append(resolvable, en)
		}
	}

	d := e.startDaemon(t, e.writeConfig(t, e.db, e.city))

	t.Run("CanonicalSelfcheck", func(t *testing.T) { e.checkSelfcheck(t, len(entries)) })
	t.Run("HTTPCanonicalSample", func(t *testing.T) { e.checkCanonicalHTTP(t, resolvable) })
	t.Run("RealEДДСQueries", func(t *testing.T) { e.checkRealQueries(t) })
	t.Run("CityAliases", func(t *testing.T) { e.checkCityAliases(t, resolvable) })
	t.Run("StrictHouseIdentity", func(t *testing.T) { e.checkStrictHouse(t, resolvable) })
	t.Run("InputLimits", func(t *testing.T) { e.checkInputLimits(t) })
	t.Run("LoadAndSafety", func(t *testing.T) { e.checkLoad(t, resolvable, d) })

	// The service must shut down cleanly on SIGTERM.
	if err := d.stop(t); err != nil {
		t.Fatalf("graceful shutdown: %v", err)
	}
}

func TestRejectsIncompatibleDeployments(t *testing.T) {
	registry := os.Getenv("ADDRKIT_E2E_REGISTRY")
	if registry == "" {
		t.Skip("ADDRKIT_E2E_REGISTRY is not set")
	}
	e := newEnv(t)
	// The binary must be built even when the fixtures below are synthetic.
	e.db = filepath.Join(t.TempDir(), "missing.db")

	t.Run("OldSchema", func(t *testing.T) {
		path := filepath.Join(t.TempDir(), "old.db")
		db, err := sql.Open("sqlite", "file:"+path)
		if err != nil {
			t.Fatalf("open: %v", err)
		}
		if _, err := db.Exec(`CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT NOT NULL)`); err != nil {
			t.Fatalf("create meta: %v", err)
		}
		if _, err := db.Exec(`INSERT INTO meta(key, value) VALUES ('schema_version', '1')`); err != nil {
			t.Fatalf("insert schema: %v", err)
		}
		db.Close()

		code, logs := e.runDaemonExpectFailure(t, e.writeConfig(t, path, e.city))
		if code == 0 {
			t.Fatalf("daemon accepted a schema-v1 database")
		}
		if !strings.Contains(logs, "unsupported index schema") {
			t.Fatalf("unexpected failure reason:\n%s", tail(logs, 20))
		}
	})

	t.Run("UnknownConfigField", func(t *testing.T) {
		cfg := e.writeConfig(t, e.db, e.city)
		raw, err := os.ReadFile(cfg)
		if err != nil {
			t.Fatalf("read config: %v", err)
		}
		var m map[string]any
		if err := json.Unmarshal(raw, &m); err != nil {
			t.Fatalf("decode config: %v", err)
		}
		m["portal"] = map[string]string{"url": "http://example"}
		out, err := json.Marshal(m)
		if err != nil {
			t.Fatalf("encode config: %v", err)
		}
		bad := filepath.Join(t.TempDir(), "config.json")
		if err := os.WriteFile(bad, out, 0o600); err != nil {
			t.Fatalf("write config: %v", err)
		}

		code, logs := e.runDaemonExpectFailure(t, bad)
		if code == 0 {
			t.Fatalf("daemon accepted an unknown config field")
		}
		if !strings.Contains(logs, "unknown field") {
			t.Fatalf("unexpected failure reason:\n%s", tail(logs, 20))
		}
	})

	t.Run("EmptyIndex", func(t *testing.T) {
		in := filepath.Join(t.TempDir(), "empty.ndjson")
		if err := os.WriteFile(in, []byte("{\"address\":\"!!!\"}\n"), 0o600); err != nil {
			t.Fatalf("write input: %v", err)
		}
		out := filepath.Join(t.TempDir(), "empty.db")
		cmd := exec.Command(e.addrkit, "seed", "-in", in, "-out", out, "-city", e.city)
		var logs bytes.Buffer
		cmd.Stdout = &logs
		cmd.Stderr = &logs
		if err := cmd.Run(); err == nil {
			t.Fatalf("seed accepted an unparseable dump")
		}
		if !strings.Contains(logs.String(), "no parseable entries") {
			t.Fatalf("unexpected failure reason:\n%s", tail(logs.String(), 20))
		}
	})
}

func newEnv(t *testing.T) *env {
	t.Helper()
	root := repoRoot(t)
	binDir := t.TempDir()
	e := &env{
		root:       root,
		binDir:     binDir,
		addrkit:    filepath.Join(binDir, "addrkit"),
		addrkitd:   filepath.Join(binDir, "addrkitd"),
		city:       envOr("ADDRKIT_E2E_CITY", defaultCity),
		httpClient: &http.Client{Timeout: 10 * time.Second},
	}
	e.parser = address.NewParser(address.NewDictionary([]address.City{{
		Name:    e.city,
		Aliases: []string{"нск", "новосиб", "новосибирская область", "нсо"},
	}}), e.city)

	e.build(t)
	return e
}

func repoRoot(t *testing.T) string {
	t.Helper()
	_, file, _, ok := runtime.Caller(0)
	if !ok {
		t.Fatal("runtime.Caller failed")
	}
	return filepath.Clean(filepath.Join(filepath.Dir(file), "..", ".."))
}

func (e *env) build(t *testing.T) {
	t.Helper()
	for _, target := range []struct{ out, pkg string }{
		{e.addrkit, "./cmd/addrkit"},
		{e.addrkitd, "./cmd/addrkitd"},
	} {
		cmd := exec.Command("go", "build", "-trimpath", "-o", target.out, target.pkg)
		cmd.Dir = e.root
		if out, err := cmd.CombinedOutput(); err != nil {
			t.Fatalf("build %s: %v\n%s", target.pkg, err, out)
		}
	}
}

func (e *env) seed(t *testing.T, registry string) []address.Entry {
	t.Helper()
	e.db = filepath.Join(t.TempDir(), "registry.db")
	cmd := exec.Command(e.addrkit, "seed",
		"-in", registry, "-out", e.db, "-version", "e2e-prod", "-city", e.city)
	out, err := cmd.CombinedOutput()
	if err != nil {
		t.Fatalf("seed: %v\n%s", err, tail(string(out), 30))
	}
	var report seedOutput
	if err := json.Unmarshal(out, &report); err != nil {
		t.Fatalf("decode seed output: %v\n%s", err, tail(string(out), 10))
	}
	e.seedReport = &report
	if report.Stats.Parsed == 0 {
		t.Fatalf("seed produced no parsed rows")
	}
	_, entries, err := registrystore.Load(e.db)
	if err != nil {
		t.Fatalf("load seeded db: %v", err)
	}
	return entries
}

func (e *env) writeConfig(t *testing.T, dbPath, city string) string {
	t.Helper()
	cfg := serviceConfig{
		Addr:        e.freeAddr(t),
		DataPath:    dbPath,
		LogLevel:    "error", // access logs would drown the e2e output
		Cities:      []cityConfig{{Name: city, Aliases: []string{"нск", "новосиб", "новосибирская область", "нсо"}}},
		DefaultCity: city,
		Resolve: resolveConfig{
			FuzzyThreshold:  0.82,
			AmbiguityMargin: 0.06,
			MaxCandidates:   5,
		},
		LLM: llmConfig{Enabled: false},
	}
	path := filepath.Join(t.TempDir(), "config.json")
	writeJSON(t, path, cfg)
	return path
}

func (e *env) freeAddr(t *testing.T) string {
	t.Helper()
	l, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatalf("free port: %v", err)
	}
	addr := l.Addr().String()
	l.Close()
	e.addr = addr
	return addr
}

func (e *env) startDaemon(t *testing.T, cfgPath string) *daemon {
	t.Helper()
	d := startProcess(t, e.addrkitd, "-config", cfgPath)
	t.Cleanup(func() { _ = d.stop(t) })

	deadline := time.Now().Add(30 * time.Second)
	for time.Now().Before(deadline) {
		if d.exited() {
			t.Fatalf("daemon exited during startup:\n%s", tail(d.logs.String(), 30))
		}
		resp, err := e.httpClient.Get("http://" + e.addr + "/v1/readyz")
		if err == nil {
			resp.Body.Close()
			if resp.StatusCode == http.StatusOK {
				return d
			}
		}
		time.Sleep(50 * time.Millisecond)
	}
	t.Fatalf("daemon did not become ready:\n%s", tail(d.logs.String(), 30))
	return nil
}

func (e *env) runDaemonExpectFailure(t *testing.T, cfgPath string) (int, string) {
	t.Helper()
	d := startProcess(t, e.addrkitd, "-config", cfgPath)
	err, ok := d.wait(15 * time.Second)
	if !ok {
		d.cmd.Process.Kill()
		t.Fatalf("daemon did not exit:\n%s", tail(d.logs.String(), 30))
	}
	code := 0
	if err != nil {
		if ee, ok := err.(*exec.ExitError); ok {
			code = ee.ExitCode()
		} else {
			code = -1
		}
	}
	return code, d.logs.String()
}

func (e *env) checkSelfcheck(t *testing.T, wantEntries int) {
	t.Helper()
	cmd := exec.Command(e.addrkit, "selfcheck", "-db", e.db)
	out, err := cmd.CombinedOutput()
	if err != nil {
		t.Fatalf("selfcheck: %v\n%s", err, tail(string(out), 20))
	}
	var rep selfcheckReport
	if err := json.Unmarshal(out, &rep); err != nil {
		t.Fatalf("decode selfcheck: %v", err)
	}
	t.Logf("selfcheck: entries=%d exact=%d rate=%.6f non_exact=%v failures=%d",
		rep.Entries, rep.Exact, rep.ExactRate, rep.NonExact, len(rep.Failures))
	if rep.Entries != wantEntries {
		t.Errorf("selfcheck entries = %d, want %d", rep.Entries, wantEntries)
	}
	if rep.ExactRate < 0.999 {
		t.Errorf("exact rate = %.6f, want >= 0.999", rep.ExactRate)
	}
	if rep.NonExact["not_found"] > 0 || rep.NonExact["ambiguous"] > 0 {
		t.Errorf("canonical rows must not be not_found/ambiguous: %v", rep.NonExact)
	}
	if len(rep.Failures) > 20 {
		t.Errorf("too many known-corrupt source rows: %d", len(rep.Failures))
	}
}

func (e *env) checkCanonicalHTTP(t *testing.T, entries []address.Entry) {
	t.Helper()
	n := envInt("ADDRKIT_E2E_SAMPLE", 2000)
	if n > len(entries) {
		n = len(entries)
	}
	idx := sampleIndexes(len(entries), n, 7)

	var mu sync.Mutex
	var firstErr error
	setErr := func(err error) {
		mu.Lock()
		if firstErr == nil {
			firstErr = err
		}
		mu.Unlock()
	}
	runConcurrent(32, len(idx), func(i int) {
		want := entries[idx[i]]
		code, body, err := e.do("/v1/resolve", map[string]string{"address": want.Address})
		if err != nil {
			setErr(fmt.Errorf("%q: %v", want.Address, err))
			return
		}
		if code != http.StatusOK {
			setErr(fmt.Errorf("%q: HTTP %d: %s", want.Address, code, tail(string(body), 3)))
			return
		}
		var res match.Result
		if err := json.Unmarshal(body, &res); err != nil {
			setErr(fmt.Errorf("%q: decode: %v", want.Address, err))
			return
		}
		if res.Status != address.StatusFound || res.Address != want.Address || res.City != e.city {
			setErr(fmt.Errorf("%q: got status=%s address=%q city=%q", want.Address, res.Status, res.Address, res.City))
		}
	})
	mu.Lock()
	err := firstErr
	mu.Unlock()
	if err != nil {
		t.Fatal(err)
	}
	t.Logf("canonical HTTP sample: %d/%d exact", n, n)
}

func (e *env) checkRealQueries(t *testing.T) {
	t.Helper()
	path := os.Getenv("ADDRKIT_E2E_LOG")
	if path == "" {
		t.Skip("ADDRKIT_E2E_LOG is not set")
	}
	logs, err := readAddressLog(path)
	if err != nil {
		t.Fatalf("read address log: %v", err)
	}
	if len(logs) == 0 {
		t.Skip("address log is empty")
	}

	// Canonical requests are what the old pipeline passed to Elasticsearch:
	// addrkit must handle all of them.
	seen := map[string]struct{}{}
	var canonical []string
	for _, rec := range logs {
		if rec.Request == "" {
			continue
		}
		if _, ok := seen[rec.Request]; ok {
			continue
		}
		seen[rec.Request] = struct{}{}
		canonical = append(canonical, rec.Request)
	}
	for _, q := range canonical {
		code, body := e.post(t, "/v1/resolve", map[string]string{"address": q})
		res := match.Result{}
		_ = json.Unmarshal(body, &res)
		if code != http.StatusOK || res.Status != address.StatusFound {
			t.Errorf("canonical EДДС request %q: HTTP %d status=%s", q, code, res.Status)
		}
	}

	// Raw utterances also contain spelled-out numbers and typos which the old
	// pipeline fixed with SEE before calling the normalizer. They are
	// informational, but the found rate must not collapse.
	counts := map[address.Status]int{}
	for _, rec := range logs {
		if rec.Utterance == "" {
			continue
		}
		_, body := e.post(t, "/v1/resolve", map[string]string{"address": rec.Utterance})
		var res match.Result
		_ = json.Unmarshal(body, &res)
		counts[res.Status]++
	}
	total := 0
	for _, n := range counts {
		total += n
	}
	foundRate := float64(counts[address.StatusFound]) / float64(max(total, 1))
	t.Logf("real EДДС queries: canonical=%d raw=%d statuses=%v found_rate=%.2f",
		len(canonical), total, counts, foundRate)
	if foundRate < 0.5 {
		t.Errorf("raw utterance found rate = %.2f, want >= 0.50", foundRate)
	}
}

func (e *env) checkCityAliases(t *testing.T, entries []address.Entry) {
	t.Helper()
	base := entries[0].Address
	for _, prefix := range []string{"нск, ", "г. Новосибирск, ", "Новосибирская область, ", "НСО, "} {
		code, body := e.post(t, "/v1/resolve", map[string]string{"address": prefix + base})
		var res match.Result
		_ = json.Unmarshal(body, &res)
		if code != http.StatusOK || res.Status != address.StatusFound || res.Address != base {
			t.Errorf("%q prefix: HTTP %d status=%s address=%q", strings.TrimSpace(prefix), code, res.Status, res.Address)
		}
	}

	code, body := e.post(t, "/v1/resolve", map[string]string{"address": base, "city": e.city})
	var res match.Result
	_ = json.Unmarshal(body, &res)
	if code != http.StatusOK || res.Status != address.StatusFound || res.City != e.city {
		t.Errorf("city hint: HTTP %d status=%s city=%q", code, res.Status, res.City)
	}

	for _, q := range []string{"нск", "г. Новосибирск", "Новосибирская область", "НСО"} {
		code, body := e.post(t, "/v1/city", map[string]string{"query": q})
		var got struct {
			Status string `json:"status"`
			City   string `json:"city"`
		}
		_ = json.Unmarshal(body, &got)
		if code != http.StatusOK || got.Status != "found" || got.City != e.city {
			t.Errorf("/v1/city %q: HTTP %d %+v", q, code, got)
		}
	}

	code, body = e.get(t, "/v1/cities")
	var cities struct {
		Cities []string `json:"cities"`
	}
	_ = json.Unmarshal(body, &cities)
	if code != http.StatusOK || len(cities.Cities) != 1 || cities.Cities[0] != e.city {
		t.Errorf("/v1/cities: HTTP %d %+v", code, cities)
	}

	code, body = e.post(t, "/v1/resolve", map[string]string{"address": "Томск, " + base})
	_ = json.Unmarshal(body, &res)
	if res.Status == address.StatusFound {
		t.Errorf("unknown city must not resolve: %+v", res)
	}
}

func (e *env) checkStrictHouse(t *testing.T, entries []address.Entry) {
	t.Helper()
	withStructure := firstWhere(entries, func(en address.Entry) bool { return en.Structure != "" })
	if withStructure == nil {
		t.Fatal("no entry with structure in the registry")
	}
	if res := e.resolve(t, withStructure.Address); res.Status != address.StatusFound || res.Address != withStructure.Address {
		t.Fatalf("exact structure: %+v", res)
	}
	wrong := strings.Replace(withStructure.Address, "стр. "+withStructure.Structure, "стр. 999", 1)
	if wrong == withStructure.Address {
		wrong = withStructure.Address + " стр. 999"
	}
	if res := e.resolve(t, wrong); res.Status == address.StatusFound {
		t.Errorf("wrong structure resolved to %q", res.Address)
	}

	withHouse := firstWhere(entries, func(en address.Entry) bool {
		return en.House != "" && en.HouseFrac == "" && en.HouseLetter == ""
	})
	if withHouse == nil {
		t.Fatal("no entry with a house in the registry")
	}
	badHouse := replaceHouse(withHouse.Address, withHouse.House, "99999")
	if res := e.resolve(t, badHouse); res.Status == address.StatusFound {
		t.Errorf("wrong house resolved to %q", res.Address)
	}

	withBuilding := firstWhere(entries, func(en address.Entry) bool { return en.Building != "" })
	if withBuilding != nil {
		bad := strings.Replace(withBuilding.Address, "к"+withBuilding.Building, "к999", 1)
		if res := e.resolve(t, bad); res.Status == address.StatusFound {
			t.Errorf("wrong building resolved to %q", res.Address)
		}
	}
}

func (e *env) checkInputLimits(t *testing.T) {
	t.Helper()
	long := strings.Repeat("а", 600)
	if code, _ := e.post(t, "/v1/resolve", map[string]string{"address": long}); code != http.StatusBadRequest {
		t.Errorf("oversized address: HTTP %d, want 400", code)
	}
	if code, _ := e.post(t, "/v1/resolve", map[string]string{}); code != http.StatusBadRequest {
		t.Errorf("empty address: HTTP %d, want 400", code)
	}
	if code, _ := e.post(t, "/v1/resolve", "not-json"); code != http.StatusBadRequest {
		t.Errorf("invalid json: HTTP %d, want 400", code)
	}
	if code, _ := e.post(t, "/v1/city", map[string]string{}); code != http.StatusBadRequest {
		t.Errorf("empty city query: HTTP %d, want 400", code)
	}
}

type loadStats struct {
	mu          sync.Mutex
	latencies   []time.Duration
	canonical   int
	canonicalOK int
	safety      int
	httpErrors  int
	found       int
	statuses    map[address.Status]int
}

func (s *loadStats) record(lat time.Duration, canonical bool, res match.Result, err error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.latencies = append(s.latencies, lat)
	if canonical {
		s.canonical++
	}
	if err != nil {
		s.httpErrors++
		return
	}
	if res.Status == address.StatusFound {
		s.found++
	}
	if s.statuses == nil {
		s.statuses = map[address.Status]int{}
	}
	s.statuses[res.Status]++
	if canonical && res.Status == address.StatusFound {
		s.canonicalOK++
	}
}

func (e *env) checkLoad(t *testing.T, entries []address.Entry, d *daemon) {
	t.Helper()
	total := envInt("ADDRKIT_E2E_LOAD", 10_000)
	workers := 64
	rng := rand.New(rand.NewSource(42))

	type job struct {
		query     string
		canonical bool
		want      address.Entry
	}
	jobs := make([]job, total)
	for i := range jobs {
		en := entries[rng.Intn(len(entries))]
		q := en.Address
		canonical := true
		switch {
		case i%10 == 9: // 10% typos
			q = insertTypo(q, rng)
			canonical = false
		case i%17 == 0: // every 17th query carries a city alias
			q = "нск, " + q
		}
		jobs[i] = job{query: q, canonical: canonical, want: en}
	}

	stats := &loadStats{latencies: make([]time.Duration, 0, total)}
	runConcurrent(workers, len(jobs), func(i int) {
		j := jobs[i]
		start := time.Now()
		code, body, reqErr := e.do("/v1/resolve", map[string]string{"address": j.query})
		lat := time.Since(start)
		var res match.Result
		err := reqErr
		if err == nil && code != http.StatusOK {
			err = fmt.Errorf("HTTP %d", code)
		}
		if err == nil {
			err = json.Unmarshal(body, &res)
		}
		if err == nil {
			if verr := e.verifySafety(j.query, res); verr != nil {
				stats.mu.Lock()
				stats.safety++
				stats.mu.Unlock()
				t.Errorf("safety violation: %v", verr)
			}
		}
		stats.record(lat, j.canonical, res, err)
	})

	stats.mu.Lock()
	defer stats.mu.Unlock()
	sort.Slice(stats.latencies, func(i, j int) bool { return stats.latencies[i] < stats.latencies[j] })
	p := func(q float64) time.Duration {
		if len(stats.latencies) == 0 {
			return 0
		}
		i := int(float64(len(stats.latencies)-1) * q)
		return stats.latencies[i]
	}
	rss := processRSS(d.cmd.Process.Pid)
	t.Logf("load: n=%d canonical=%d canonical_found=%d safety=%d http_errors=%d statuses=%v p50=%s p95=%s p99=%s rss=%dMB",
		total, stats.canonical, stats.canonicalOK, stats.safety, stats.httpErrors, stats.statuses,
		p(0.50), p(0.95), p(0.99), rss/(1<<20))

	if stats.httpErrors > 0 {
		t.Errorf("HTTP errors: %d", stats.httpErrors)
	}
	if stats.canonicalOK != stats.canonical {
		t.Errorf("canonical found = %d/%d", stats.canonicalOK, stats.canonical)
	}
	if stats.safety > 0 {
		t.Errorf("safety violations: %d", stats.safety)
	}
	if stats.found == 0 {
		t.Error("no found results during load")
	}
	if p(0.99) > maxP99 {
		t.Errorf("p99 latency = %s, want <= %s", p(0.99), maxP99)
	}
	if rss > maxRSS {
		t.Errorf("RSS = %d MB, want <= %d MB", rss/(1<<20), maxRSS/(1<<20))
	}
}

// verifySafety enforces the core promise: a "found" answer must be a real
// registry row and must not change the requested house unit.
func (e *env) verifySafety(query string, res match.Result) error {
	if res.Status != address.StatusFound {
		return nil
	}
	qp := e.parser.ParseInCity(query, e.city)
	rp := e.parser.ParseInCity(res.Address, e.city)
	if qp.HasHouse && rp.HasHouse {
		if qp.House != rp.House || qp.HouseFrac != rp.HouseFrac || qp.HouseLetter != rp.HouseLetter ||
			qp.Building != rp.Building || qp.BuildingFrac != rp.BuildingFrac || qp.BuildingLetter != rp.BuildingLetter ||
			qp.Structure != rp.Structure {
			return fmt.Errorf("%q -> %q changes the house unit", query, res.Address)
		}
	}
	return nil
}

func (e *env) resolve(t *testing.T, q string) match.Result {
	t.Helper()
	code, body := e.post(t, "/v1/resolve", map[string]string{"address": q})
	if code != http.StatusOK {
		t.Fatalf("resolve %q: HTTP %d", q, code)
	}
	var res match.Result
	if err := json.Unmarshal(body, &res); err != nil {
		t.Fatalf("decode: %v", err)
	}
	return res
}

func (e *env) post(t *testing.T, path string, payload any) (int, []byte) {
	t.Helper()
	code, body, err := e.do(path, payload)
	if err != nil {
		t.Fatalf("POST %s: %v", path, err)
	}
	return code, body
}

// do performs an HTTP request without touching *testing.T, so it is safe to
// call from worker goroutines.
func (e *env) do(path string, payload any) (int, []byte, error) {
	var r io.Reader
	switch v := payload.(type) {
	case string:
		r = strings.NewReader(v)
	default:
		b, err := json.Marshal(v)
		if err != nil {
			return 0, nil, err
		}
		r = bytes.NewReader(b)
	}
	resp, err := e.httpClient.Post("http://"+e.addr+path, "application/json", r)
	if err != nil {
		return 0, nil, err
	}
	defer resp.Body.Close()
	body, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
	return resp.StatusCode, body, nil
}

func (e *env) get(t *testing.T, path string) (int, []byte) {
	t.Helper()
	resp, err := e.httpClient.Get("http://" + e.addr + path)
	if err != nil {
		return 0, nil
	}
	defer resp.Body.Close()
	body, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
	return resp.StatusCode, body
}

type daemon struct {
	cmd      *exec.Cmd
	logs     *lockedBuffer
	finished chan struct{}
	once     sync.Once
	err      error
}

func startProcess(t *testing.T, name string, args ...string) *daemon {
	t.Helper()
	cmd := exec.Command(name, args...)
	logs := &lockedBuffer{}
	cmd.Stdout = logs
	cmd.Stderr = logs
	if err := cmd.Start(); err != nil {
		t.Fatalf("start %s: %v", name, err)
	}
	d := &daemon{cmd: cmd, logs: logs, finished: make(chan struct{})}
	go func() {
		err := cmd.Wait()
		d.once.Do(func() {
			d.err = err
			close(d.finished)
		})
	}()
	return d
}

func (d *daemon) exited() bool {
	select {
	case <-d.finished:
		return true
	default:
		return false
	}
}

func (d *daemon) wait(timeout time.Duration) (error, bool) {
	select {
	case <-d.finished:
		return d.err, true
	case <-time.After(timeout):
		return nil, false
	}
}

// stop asks the daemon to terminate and returns its exit status. Calling it
// twice is safe; a daemon that already exited is not signalled again.
func (d *daemon) stop(t *testing.T) error {
	t.Helper()
	if !d.exited() {
		_ = d.cmd.Process.Signal(syscall.SIGTERM)
	}
	err, ok := d.wait(10 * time.Second)
	if !ok {
		_ = d.cmd.Process.Kill()
		err, _ = d.wait(5 * time.Second)
	}
	return err
}

type lockedBuffer struct {
	mu  sync.Mutex
	buf bytes.Buffer
}

func (b *lockedBuffer) Write(p []byte) (int, error) {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.buf.Write(p)
}

func (b *lockedBuffer) String() string {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.buf.String()
}

type addressLogRecord struct {
	Utterance string
	Request   string
}

// readAddressLog reads the EДДС address log (bulk NDJSON) and extracts the
// canonical request and the raw user utterance. It deliberately does not keep
// dialog ids or phone numbers.
func readAddressLog(path string) ([]addressLogRecord, error) {
	f, err := os.Open(path)
	if err != nil {
		return nil, err
	}
	defer f.Close()

	var out []addressLogRecord
	dec := json.NewDecoder(f)
	for {
		var raw json.RawMessage
		if err := dec.Decode(&raw); err != nil {
			if err == io.EOF {
				break
			}
			return nil, err
		}
		var obj map[string]json.RawMessage
		if err := json.Unmarshal(raw, &obj); err != nil {
			continue
		}
		if _, ok := obj["index"]; ok {
			continue // bulk action line
		}
		var rec struct {
			Address string `json:"address"`
			Request string `json:"request"`
		}
		_ = json.Unmarshal(raw, &rec)
		item := addressLogRecord{Utterance: strings.TrimSpace(rec.Address)}
		var req struct {
			Address string `json:"address"`
		}
		if json.Unmarshal([]byte(rec.Request), &req) == nil {
			item.Request = strings.TrimSpace(req.Address)
		}
		out = append(out, item)
	}
	return out, nil
}

func runConcurrent(workers, n int, fn func(i int)) {
	var wg sync.WaitGroup
	ch := make(chan int)
	for w := 0; w < workers; w++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for i := range ch {
				fn(i)
			}
		}()
	}
	for i := 0; i < n; i++ {
		ch <- i
	}
	close(ch)
	wg.Wait()
}

func sampleIndexes(n, sample int, seed int64) []int {
	rng := rand.New(rand.NewSource(seed))
	perm := rng.Perm(n)
	out := make([]int, sample)
	copy(out, perm[:sample])
	return out
}

func firstWhere(entries []address.Entry, pred func(address.Entry) bool) *address.Entry {
	for i := range entries {
		if pred(entries[i]) {
			return &entries[i]
		}
	}
	return nil
}

func replaceHouse(addressText, old, new string) string {
	if old == "" {
		return addressText
	}
	// Replace the house token only at the end of the street phrase.
	idx := strings.LastIndex(addressText, " "+old)
	if idx < 0 {
		return addressText
	}
	return addressText[:idx] + " " + new + addressText[idx+1+len(old):]
}

// insertTypo swaps two adjacent runes in the longest Cyrillic word, emulating a
// citizen typo without changing the house number.
func insertTypo(s string, rng *rand.Rand) string {
	tokens := strings.Fields(s)
	best := -1
	for i, tok := range tokens {
		if !isCyrillic(tok) || len([]rune(tok)) < 5 {
			continue
		}
		if best < 0 || len([]rune(tok)) > len([]rune(tokens[best])) {
			best = i
		}
	}
	if best < 0 {
		return s
	}
	r := []rune(tokens[best])
	i := 1 + rng.Intn(len(r)-2)
	r[i], r[i+1] = r[i+1], r[i]
	tokens[best] = string(r)
	return strings.Join(tokens, " ")
}

func isCyrillic(s string) bool {
	for _, r := range s {
		if r < 'а' || r > 'я' {
			return false
		}
	}
	return len(s) > 0
}

func processRSS(pid int) int64 {
	data, err := os.ReadFile(fmt.Sprintf("/proc/%d/status", pid))
	if err != nil {
		return 0
	}
	for _, line := range strings.Split(string(data), "\n") {
		if strings.HasPrefix(line, "VmRSS:") {
			var kb int64
			fmt.Sscanf(strings.TrimSpace(strings.TrimPrefix(line, "VmRSS:")), "%d", &kb)
			return kb * 1024
		}
	}
	return 0
}

func writeJSON(t *testing.T, path string, v any) {
	t.Helper()
	data, err := json.MarshalIndent(v, "", "  ")
	if err != nil {
		t.Fatalf("marshal json: %v", err)
	}
	if err := os.WriteFile(path, data, 0o600); err != nil {
		t.Fatalf("write %s: %v", path, err)
	}
}

func envOr(name, fallback string) string {
	if v := strings.TrimSpace(os.Getenv(name)); v != "" {
		return v
	}
	return fallback
}

func envInt(name string, fallback int) int {
	v := strings.TrimSpace(os.Getenv(name))
	if v == "" {
		return fallback
	}
	var n int
	if _, err := fmt.Sscanf(v, "%d", &n); err != nil || n <= 0 {
		return fallback
	}
	return n
}

func tail(s string, n int) string {
	lines := strings.Split(strings.TrimSpace(s), "\n")
	if len(lines) > n {
		lines = lines[len(lines)-n:]
	}
	return strings.Join(lines, "\n")
}