munin/last_test.go

//go:build last

// Lasttest: viele Agenten greifen gleichzeitig über echtes HTTP und MCP zu.
//
// Er läuft NICHT bei jedem "go test ./...", weil er Zeit braucht und die
// Maschine auslastet. Das Build-Tag oben schaltet ihn erst ein, wenn man
// ausdrücklich danach fragt:
//
//	go test -tags last -run TestLast -v .
//
// Mit eigenen Werten (alle Schalter siehe die flag.-Zeilen unten):
//
//	go test -tags last -run TestLast -v . -last.agenten=64 -last.dauer=30s
//
// Drei Aufstellungen mit derselben Zahl Agenten, damit die Zahlen vergleichbar
// sind:
//
//   - ein Projekt: alle Agenten im selben Projekt
//   - ein Workspace, mehrere Projekte: teilen sich eine Datei und damit die
//     eine Schreibverbindung (siehe openPool in internal/store/store.go)
//   - mehrere Workspaces: je eine eigene Datei, schreiben echt parallel
//
// Am Ende prüft jede Aufstellung, ob nichts verloren ging und ob der
// Suchindex zum Stand der Datenbank passt.
package main

import (
	"context"
	"flag"
	"fmt"
	"io"
	"log"
	"log/slog"
	"math/rand/v2"
	"net/http"
	"net/http/httptest"
	"path/filepath"
	"slices"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/modelcontextprotocol/go-sdk/mcp"

	"munin/internal/auth"
	"munin/internal/search"
	"munin/internal/store"
	"munin/internal/workspace"
)

var (
	lastAgents   = flag.Int("last.agenten", 32, "Zahl der gleichzeitigen Agenten je Aufstellung")
	lastDuration = flag.Duration("last.dauer", 10*time.Second, "Laufzeit je Aufstellung")
	lastProjects = flag.Int("last.projekte", 4, "Projekte bzw. Workspaces in den Aufstellungen mit mehreren")
	lastEntries  = flag.Int("last.eintraege", 500, "Einträge, die jedes Projekt vorab bekommt")
	lastSlugs    = flag.Int("last.slugs", 20, "Bezeichner, auf die die Agenten je Projekt schreiben — klein heißt viel Gedränge auf denselben Einträgen")
	lastWrites   = flag.Int("last.schreibanteil", 30, "Anteil schreibender Aufrufe in Prozent, der Rest ist recall und get")
)

func TestLast(t *testing.T) {
	// Jede Anfrage würde sonst eine Logzeile schreiben — bei zehntausenden
	// Aufrufen geht die Auswertung darin unter.
	log.SetOutput(io.Discard)
	slog.SetDefault(slog.New(slog.NewTextHandler(io.Discard, nil)))

	n := *lastProjects
	setups := []struct {
		name                 string
		workspaces, projects int // projekte = je Workspace
	}{
		{"ein Projekt", 1, 1},
		{fmt.Sprintf("ein Workspace, %d Projekte", n), 1, n},
		{fmt.Sprintf("%d Workspaces, je 1 Projekt", n), n, 1},
	}

	for _, a := range setups {
		t.Run(a.name, func(t *testing.T) {
			run(t, a.workspaces, a.projects)
		})
	}
}

// target ist ein Projekt samt den Keys, mit denen Agenten es ansprechen.
type target struct {
	ws      *workspace.Workspace
	project store.Project
	keys    []string
}

// run baut eine Aufstellung auf, lässt die Agenten laufen und wertet aus.
func run(t *testing.T, numWorkspaces, projectsPerWorkspace int) {
	ctx := context.Background()
	dir := t.TempDir()

	control, err := store.OpenControl(filepath.Join(dir, "control.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { control.Close() })
	workspaces := workspace.NewManager(store.NewSQLite("", dir))
	t.Cleanup(func() { workspaces.Close() })

	// Dieselbe Verdrahtung wie im Betrieb. Schreibbudget 0 = unbegrenzt — es
	// soll eine Schleife stoppen und würde hier die Messung abwürgen.
	handler, err := buildHandler(control, workspaces, nil, config{
		sessionSecret:    strings.Repeat("x", 32),
		adminToken:       strings.Repeat("y", 32),
		maxAPIBody:       defaultMaxAPIBodyBytes,
		writeLimitHourly: 0,
	})
	if err != nil {
		t.Fatal(err)
	}
	server := httptest.NewServer(handler)
	t.Cleanup(server.Close)

	// ── Daten anlegen ────────────────────────────────────────────────────────
	agents := *lastAgents
	numTargets := numWorkspaces * projectsPerWorkspace
	var targets []*target

	for w := 0; w < numWorkspaces; w++ {
		wsEntry, err := control.CreateWorkspace(ctx, fmt.Sprintf("Workspace %d", w))
		if err != nil {
			t.Fatal(err)
		}
		ws, err := workspaces.Get(wsEntry.ID)
		if err != nil {
			t.Fatal(err)
		}
		for p := 0; p < projectsPerWorkspace; p++ {
			project, err := ws.Store.CreateProject(ctx, wsEntry.ID, fmt.Sprintf("projekt-%d", p), "Projekt", "")
			if err != nil {
				t.Fatal(err)
			}
			z := &target{ws: ws, project: project}
			targets = append(targets, z)

			for i := 0; i < *lastEntries; i++ {
				if _, err := ws.Store.PutEntry(ctx, store.EntryWrite{
					ProjectID: project.ID, Slug: fmt.Sprintf("vorab-%04d", i), Title: fmt.Sprintf("Vorab %d", i),
					Content: fillerText(i), Source: store.SourceHuman,
				}); err != nil {
					t.Fatal(err)
				}
			}
		}
	}

	// Jeder Agent bekommt einen eigenen Key, wie ein echter Nutzer.
	for a := 0; a < agents; a++ {
		z := targets[a%numTargets]
		plaintext, err := auth.GenerateKey()
		if err != nil {
			t.Fatal(err)
		}
		hash := auth.HashKey(plaintext)
		if _, err := z.ws.Store.CreateAPIKey(ctx, z.project.ID, fmt.Sprintf("agent-%d", a), hash, string(auth.RoleWrite)); err != nil {
			t.Fatal(err)
		}
		if err := control.IndexKey(ctx, hash, z.ws.ID); err != nil {
			t.Fatal(err)
		}
		z.keys = append(z.keys, plaintext)
	}

	// ── Agenten laufen lassen ────────────────────────────────────────────────
	// Ein gemeinsamer Transport, der je Agent eine Verbindung offen hält. Mit
	// den Vorgaben von Go (2 offene Verbindungen je Ziel) öffnen die Agenten
	// sonst laufend neue, bis dem Testrechner die lokalen Ports ausgehen
	// ("can't assign requested address"). Echte Agenten sitzen auf
	// verschiedenen Rechnern und kennen das Problem nicht — es wäre also ein
	// Fehler der Messung, nicht des Servers.
	httpTransport := http.DefaultTransport.(*http.Transport).Clone()
	httpTransport.MaxIdleConnsPerHost = agents
	t.Cleanup(httpTransport.CloseIdleConnections)

	measurement := newMeasurement()
	deadline := time.Now().Add(*lastDuration)
	var wg sync.WaitGroup

	for a := 0; a < agents; a++ {
		z := targets[a%numTargets]
		key := z.keys[a/numTargets]
		wg.Add(1)
		go func(a int) {
			defer wg.Done()
			agent(ctx, httpTransport, server.URL, key, a, deadline, measurement)
		}(a)
	}
	wg.Wait()

	// ── Auswerten ────────────────────────────────────────────────────────────
	measurement.report(t, *lastDuration)

	if count, first := measurement.failures(); count > 0 {
		t.Errorf("%d fehlgeschlagene Aufrufe, erste: %v", count, first)
	}

	checkInventory(t, targets, measurement.successfulWrites())
}

// agent verhält sich wie ein Agent in einer Schleife: suchen, lesen,
// gelegentlich schreiben — bis die Zeit um ist.
func agent(ctx context.Context, httpTransport http.RoundTripper, url, key string, num int, deadline time.Time, m *measurement) {
	httpClient := &http.Client{
		Transport: &withKey{key: key, next: httpTransport},
		Timeout:   30 * time.Second,
	}
	transport := &mcp.StreamableClientTransport{
		Endpoint:   url + "/mcp",
		HTTPClient: httpClient,
		// Der zusätzliche Rückkanal für Nachrichten vom Server wird hier nicht
		// gebraucht und würde nur eine weitere offene Verbindung je Agent kosten.
		DisableStandaloneSSE: true,
	}
	client := mcp.NewClient(&mcp.Implementation{Name: fmt.Sprintf("agent-%d", num), Version: "1"}, nil)

	start := time.Now()
	session, err := client.Connect(ctx, transport, nil)
	m.record("verbinden", time.Since(start), err)
	if err != nil {
		return
	}
	defer session.Close()

	rng := rand.New(rand.NewPCG(uint64(num), 42))
	for i := 0; time.Now().Before(deadline); i++ {
		var tool string
		var args map[string]any

		switch w := rng.IntN(100); {
		case w < *lastWrites:
			tool = "remember"
			// Ein eindeutiges Wort je Schreibvorgang — daran erkennt
			// pruefeBestand, ob der Suchindex den letzten Stand kennt.
			//
			// force: Alle umkämpften Einträge tragen denselben Titel. Ohne
			// force hielte die Dublettenprüfung jede Neuanlage nach der
			// ersten an — der Aufruf zählte als erfolgreich, ohne dass eine
			// Fassung entsteht, und die Bilanz in pruefeBestand ginge nicht auf.
			args = map[string]any{
				"slug":    fmt.Sprintf("gemeinsam-%02d", rng.IntN(*lastSlugs)),
				"title":   "Gemeinsamer Eintrag",
				"content": "Stand " + uniqueWord(num, i) + "\n\n" + fillerText(i),
				"force":   true,
			}
		case w < *lastWrites+(100-*lastWrites)*2/3:
			tool = "recall"
			args = map[string]any{"query": searchTerms[rng.IntN(len(searchTerms))]}
		default:
			tool = "get"
			args = map[string]any{"slug": fmt.Sprintf("vorab-%04d", rng.IntN(max(1, *lastEntries)))}
		}

		start := time.Now()
		erg, err := session.CallTool(ctx, &mcp.CallToolParams{Name: tool, Arguments: args})
		if err == nil && erg.IsError {
			err = fmt.Errorf("%s: %s", tool, resultText(erg))
		}
		m.record(tool, time.Since(start), err)
	}
}

// checkInventory vergleicht am Ende Datenbank und Suchindex.
//
// Zwei Dinge müssen stimmen:
//   - Kein Schreibvorgang ging verloren: Jeder erfolgreiche remember-Aufruf
//     hat eine Fassung erzeugt. Die Summe der Fassungen aller umkämpften
//     Einträge muss also genau die Zahl der erfolgreichen Aufrufe sein.
//   - Der Suchindex kennt den letzten Stand: Das eindeutige Wort der letzten
//     Fassung muss den Eintrag finden.
func checkInventory(t *testing.T, targets []*target, succeeded int) {
	t.Helper()
	ctx := context.Background()

	totalVersions, stale := 0, 0
	for _, z := range targets {
		idx, err := z.ws.Search.For(ctx, z.project.ID)
		if err != nil {
			t.Fatal(err)
		}
		for s := 0; s < *lastSlugs; s++ {
			slug := fmt.Sprintf("gemeinsam-%02d", s)
			e, err := z.ws.Store.EntryBySlug(ctx, z.project.ID, slug)
			if err != nil {
				continue // nie beschrieben
			}
			totalVersions += e.Version

			wort := strings.Fields(e.Content)[1]
			found := slices.ContainsFunc(idx.Search(wort, 5), func(r search.Result) bool { return r.DocID == slug })
			if !found {
				stale++
			}
		}
	}

	if totalVersions != succeeded {
		t.Errorf("%d erfolgreiche remember-Aufrufe, aber %d Fassungen gespeichert", succeeded, totalVersions)
	}
	if stale > 0 {
		t.Errorf("%d Einträge: Der Suchindex kennt die letzte Fassung nicht", stale)
	}
	t.Logf("Bestand geprüft: %d Fassungen, Suchindex %s", totalVersions,
		map[bool]string{true: "aktuell", false: "VERALTET"}[stale == 0])
}

// ── Messung ──────────────────────────────────────────────────────────────────

// measurement sammelt Dauer und Fehler aller Aufrufe, getrennt nach Werkzeug.
type measurement struct {
	mu        sync.Mutex
	durations map[string][]time.Duration
	failed    map[string]int
	first     []error
}

func newMeasurement() *measurement {
	return &measurement{durations: map[string][]time.Duration{}, failed: map[string]int{}}
}

func (m *measurement) record(tool string, d time.Duration, err error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	if err != nil {
		m.failed[tool]++
		if len(m.first) < 10 {
			m.first = append(m.first, err)
		}
		return
	}
	m.durations[tool] = append(m.durations[tool], d)
}

// failures gibt die Zahl aller fehlgeschlagenen Aufrufe und die ersten davon
// zurück.
func (m *measurement) failures() (count int, first []error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	for _, n := range m.failed {
		count += n
	}
	return count, m.first
}

func (m *measurement) successfulWrites() int {
	m.mu.Lock()
	defer m.mu.Unlock()
	return len(m.durations["remember"])
}

// report schreibt eine Tabelle: Aufrufe je Sekunde und wie lange die
// typische (p50), die langsamsten 5 % (p95) und 1 % (p99) Aufrufe dauerten.
func (m *measurement) report(t *testing.T, duration time.Duration) {
	m.mu.Lock()
	defer m.mu.Unlock()

	var b strings.Builder
	fmt.Fprintf(&b, "\n%-10s %8s %8s %9s %9s %9s %9s %7s\n", "Werkzeug", "Aufrufe", "je s", "p50", "p95", "p99", "max", "Fehler")
	for _, w := range []string{"verbinden", "recall", "get", "remember"} {
		d := m.durations[w]
		slices.Sort(d)
		je := float64(len(d)) / duration.Seconds()
		if w == "verbinden" {
			je = 0 // einmal je Agent, keine Rate
		}
		fmt.Fprintf(&b, "%-10s %8d %8.0f %9s %9s %9s %9s %7d\n", w, len(d), je,
			percentile(d, 50), percentile(d, 95), percentile(d, 99), percentile(d, 100), m.failed[w])
	}
	t.Log(b.String())
}

func percentile(sorted []time.Duration, p int) string {
	if len(sorted) == 0 {
		return "-"
	}
	i := (len(sorted) - 1) * p / 100
	return sorted[i].Round(10 * time.Microsecond).String()
}

// ── Hilfen ───────────────────────────────────────────────────────────────────

// withKey hängt den Bearer-Key an jede Anfrage — so wie es ein MCP-Client mit
// "headers" in seiner Konfiguration tut.
type withKey struct {
	key  string
	next http.RoundTripper
}

func (m *withKey) RoundTrip(r *http.Request) (*http.Response, error) {
	r = r.Clone(r.Context())
	r.Header.Set("Authorization", "Bearer "+m.key)
	return m.next.RoundTrip(r)
}

func resultText(r *mcp.CallToolResult) string {
	var b strings.Builder
	for _, c := range r.Content {
		if t, ok := c.(*mcp.TextContent); ok {
			b.WriteString(t.Text)
		}
	}
	return b.String()
}

// uniqueWord bildet aus Agent und Durchlauf ein Wort nur aus Buchstaben,
// damit der Tokenizer es nicht an Ziffern zerlegt.
func uniqueWord(agent, iteration int) string {
	s := []byte(fmt.Sprintf("%04d%07d", agent, iteration))
	for i := range s {
		s[i] = 'a' + (s[i] - '0')
	}
	return "marke" + string(s)
}

var searchTerms = []string{"deployment", "server", "datenbank", "zugang", "backup", "monitoring", "release", "fehler"}

// fillerText erzeugt einen Absatz mit wechselnden Fachbegriffen, damit der
// Suchindex realistisch viele Begriffe und Treffer hat.
func fillerText(i int) string {
	var b strings.Builder
	for j := 0; j < 40; j++ {
		b.WriteString(searchTerms[(i+j*3)%len(searchTerms)])
		b.WriteString(" wird hier beschrieben, mit Details zu Schritt ")
		b.WriteString(fmt.Sprint(j))
		b.WriteString(". ")
	}
	return b.String()
}