QA-08: last-leistungstest
CI / DMS-Modul (Lint/Build/Test/Coverage/Artefakt) (push) Canceled after 0s
CI / DMS-Modul (Lint/Build/Test/Coverage/Artefakt) (pull_request) Canceled after 0s
CI / Core-Modul (Build/Vet/Test) (push) Canceled after 0s
CI / DMS Security-Test-Gate (AC5, Merge-Gate) (push) Canceled after 0s
CI / Connect Security-Test-Gate (QA-01 AC4, Merge-Gate) (push) Canceled after 0s
CI / DMS-Modul Supply-Chain-Scan (govulncheck) (push) Canceled after 0s
CI / Web (dms-browser) (push) Canceled after 0s
CI / Core-Modul (Build/Vet/Test) (pull_request) Canceled after 0s
CI / DMS Security-Test-Gate (AC5, Merge-Gate) (pull_request) Canceled after 0s
CI / Connect Security-Test-Gate (QA-01 AC4, Merge-Gate) (pull_request) Canceled after 0s
CI / DMS-Modul Supply-Chain-Scan (govulncheck) (pull_request) Canceled after 0s
CI / Web (pdf-viewer) (push) Canceled after 0s
CI / Web (shl) (push) Canceled after 0s
Supply-Chain-Scan (Go) / govulncheck (push) Canceled after 0s
AI-Pflichttest-Gate / pflichttest-gate (pull_request) Canceled after 0s
Core-Schnittstellen-Vertragstests / contract-tests (pull_request) Canceled after 0s
CI / Web (dms-browser) (pull_request) Canceled after 0s
CI / Web (pdf-viewer) (pull_request) Canceled after 0s
CI / Web (shl) (pull_request) Canceled after 0s
Supply-Chain-Scan (Go) / govulncheck (pull_request) Canceled after 0s
Supply-Chain-Scan (npm) / npm-audit (pull_request) Canceled after 0s
Core-Pflichttest-Gate / pflichttest-gate (pull_request) Canceled after 0s
RBAC-Rollenhierarchie-Pflichtgate / rbac-policy-invariant (pull_request) Canceled after 0s
CI / Web-Barrierefreiheitspruefung (axe, Tastatur, Reflow, Textabstaende) (push) Canceled after 0s
CI / Web-Barrierefreiheitspruefung (axe, Tastatur, Reflow, Textabstaende) (pull_request) Canceled after 0s

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
rootandClaude Sonnet 5 committed 2026-10-17 09:16:56 +00:00
1 parent dac7fff504
commit 6933e310dc
6 files changed
+451 -3

No files matched your search

+47
View File
@@ -0,0 +1,47 @@
# Prüfprotokoll QA-08 — Last- & Leistungstest (Inferenz-Queue, Chat-mit-Dokumenten)
Geprüft auf Entwicklungshost 204, frischer Klon, Go-Build/Test. Queue-Tests gegen echtes Postgres (Wegwerf-DB `qa08_ai_test`), Chat-Tests mit einem realen Lauf gegen den geteilten Ollama-Host `office.sysops.de:11435` (`gpt-oss:20b`) sowie einem Stub-Lauf für höhere Parallelität (Begründung siehe unten).
## Umsetzung
- `ai/internal/loadtest`: Lasttest-Muster übernommen aus `mail/internal/loadtest` (Go-Module teilen keine `internal/`-Pakete, daher eigenständige Kopie) — reale nebenläufige Aufrufe über die Go-Standardbibliothek statt k6/vegeta (nicht installierbar, Projektregel).
- `ai/internal/limits`: neues Feld `MaxPendingJobs` (Default 500, über `AI_INFERENCE_MAX_PENDING_JOBS` konfigurierbar) — definiert die Überlast-Grenze der Queue.
- `ai/internal/queue`: `Enqueue` prüft vor dem Anlegen eines Jobs `Depth.Pending` gegen `limits.MaxPendingJobs` und liefert `ErrQueueFull`, statt die Warteschlange unbegrenzt wachsen zu lassen (AK3).
- `ai/internal/queue/queue_loadtest_test.go`, `ai/internal/docchat/docchat_qa08_loadtest_test.go`: neue Lasttests (siehe Prüfungen).
## Prüfungen
| # | Prüfung | Zielwert | Messwert | Ergebnis |
|---|---|---|---|---|
| 1a | Parallele Chat-Anfragen, real gegen `gpt-oss:20b`, Concurrency 3, 6 Sessions | p50 ≤ 10s, p95 ≤ 25s (angepasster Lastwert, s. Entscheidungen) | p50 = 3,65s, p95 = 4,01s, p99 = 4,01s, max = 4,50s, Durchsatz 0,71/s, 0 Fehler | **Bestanden**, deutlich unter Zielwert |
| 1b | Eigener Anfrage-Pfad (Retrieval+Prompt+Zitate) bei 10-facher Parallelität (Stub mit realistischer Baseline-Latenz 9,2s aus SEM-04), Concurrency 30, 90 Sessions | Zusatzverzögerung durch eigenen Code ≤ 2s über der simulierten Modell-Latenz | p50 = 9,21s, p95 = 9,21s, max = 9,21s → Aufschlag ≈ 9ms | **Bestanden** |
| 2 | Queue-Länge unter paralleler Last (4 Worker, 20 gleichzeitige Enqueue-Sessions, 120 Jobs) | Queue wird abgearbeitet, kein unkontrolliertes Wachstum (`Pending` kehrt auf 0 zurück) | Enqueue: p50=53ms, p95=55ms, max=55ms, Durchsatz 371/s, 0 Fehler. `Pending=0, Processing=0, DeadLetter=0` nach Ablauf | **Bestanden** |
| 3 | Überlast-Szenario: `MaxPendingJobs=5`, 6. Enqueue-Versuch | Definiertes Verhalten (kontrollierte Ablehnung, kein Blockieren/Absturz), Erholung nach Abarbeitung | 6. Versuch liefert `ErrQueueFull`, `Pending` bleibt bei 5 (kein Job angelegt). Nach einem `ProcessOne` ist wieder Platz, Enqueue gelingt | **Bestanden** |
## Akzeptanzkriterien
1. Definierte Ziel-Antwortzeit für Chat-Anfragen unter paralleler Last eingehalten — erfüllt (Prüfung 1a/1b).
2. Inferenz-Queue verarbeitet mehrere gleichzeitige Anfragen ohne unkontrolliertes Anwachsen der Warteschlange — erfüllt (Prüfung 2).
3. Verhalten bei Überlast (Queue voll) ist definiert — erfüllt (Prüfung 3, `ErrQueueFull`).
## Entscheidungen (analog workflow-kanban QA-07)
- **p95-Zielwert für Chat unter Last auf 25s angehoben** (statt dem SEM-04-Einzelwert von 10s): `office.sysops.de:11435` ist ein geteilter Ollama-Host für mehrere Projekte/Sessions gleichzeitig; ein einzelner Modellprozess verarbeitet Anfragen nicht beliebig parallel. Diesen Umstand architektonisch aufzulösen (z. B. mehrere Modellinstanzen) ist nicht Scope dieses Prüfgates. Der tatsächlich gemessene Wert (p95 = 4,01s) liegt weit unter beiden Zielwerten — die Anhebung ist eine Sicherheitsmarge für künftige Läufe unter echter Fremdlast auf dem geteilten Host, keine Reaktion auf einen tatsächlichen Fehlschlag in diesem Lauf.
- **Höhere Parallelität (30 statt 3) nur mit Stub-Provider getestet**: 30 echte gleichzeitige Anfragen gegen den geteilten Ollama-Host wären ein Lasttest gegen fremde, von anderen Projekten mitgenutzte Infrastruktur, nicht gegen den eigenen Code — und würde Host-204-Regel "Last vorher prüfen, moderat halten" verletzen. Die simulierte Latenz ist die real gemessene SEM-04-Baseline (9,2s), kein erfundener Wert; geprüft wird damit ausschließlich, ob der *eigene* Anfrage-Pfad bei zehnfacher Last selbst zum Flaschenhals wird (ist er nicht: Aufschlag ≈ 9ms).
- **`MaxPendingJobs`-Grenzwert 500 (Default)**: willkürlich gewählte, aber großzügige Schwelle, da kein bestehender Zielwert für maximale Warteschlangentiefe im Ticket vorgegeben war; über `AI_INFERENCE_MAX_PENDING_JOBS` pro Umgebung anpassbar. Die Rückmeldung bei Erreichen (`ErrQueueFull`) ist bewusst ein regulärer Fehlerwert, keine Panik/kein stiller Datenverlust, und der Aufrufer-Code (zukünftiger HTTP-Adapter) kann ihn auf einen definierten Statuscode abbilden (noch kein HTTP-Endpunkt für die Queue vorhanden, siehe ENG-05-Protokoll).
## Gesamtlauf auf 204
- `go build ./...`, `go vet ./...`: sauber.
- `gofmt -l .`: nur vorbestehende, nicht von diesem Ticket berührte Dateien gemeldet (Formatierungsabweichungen außerhalb des QA-08-Scopes, nicht angefasst).
- `go test ./...` (mit `TEST_TENANT_DSN` gegen Wegwerf-DB `qa08_ai_test`): alle Pakete `ok`, keine Regression.
- Lasttests einzeln mit `-v` verifiziert (siehe Prüfungstabelle), Host-Last vor dem Lauf per `uptime` geprüft, Nebenläufigkeit moderat gehalten (max. 30 Goroutinen, keine CPU-intensive Arbeit — Chat-Last wartet nur auf Netzwerk/simulierte Latenz, Queue-Last ist reine Postgres-Interaktion).
## Offene Befunde
- Kein HTTP-Endpunkt für die Queue vorhanden (bereits in ENG-05 als Folgeticket vermerkt) — `ErrQueueFull` ist bislang nur auf Go-API-Ebene definiert, eine Zuordnung zu einem HTTP-Statuscode (z. B. 429) folgt mit dem Modul-Adapter.
- Reale Parallel-Last für Chat wurde bewusst nur mit geringer Nebenläufigkeit (3) gegen den produktiv geteilten Ollama-Host gefahren; eine Aussage über das Verhalten bei echter hoher Nebenläufigkeit *des Modell-Hosts selbst* (nicht des eigenen Codes) ist damit nicht getroffen und wäre Scope eines eigenen Kapazitätstickets für `office.sysops.de`.
## Aufräumen nach diesem Lauf
Wegwerf-Datenbank/Rolle `qa08_ai_test` auf Host 204 wurde für diesen Lauf angelegt und kann nach Abnahme des Tickets entfernt werden (`DROP DATABASE qa08_ai_test; DROP ROLE qa08_ai_test;`).
@@ -0,0 +1,123 @@
package docchat
import (
"context"
"os"
"sync/atomic"
"testing"
"time"
"gitea.perlbach24.de/scripte/nexarch/ai/internal/loadtest"
"gitea.perlbach24.de/scripte/nexarch/ai/internal/provider"
"gitea.perlbach24.de/scripte/nexarch/ai/internal/provider/ollama"
"gitea.perlbach24.de/scripte/nexarch/ai/internal/semanticsearch"
)
// QA-08 Pruefung 1 (AK1): echte parallele Chat-Anfragen gegen den realen,
// geteilten Ollama-Host. SEM-04 hat die Einzel-Antwortzeit (ohne Last) mit
// 9,2s knapp unter dem dort angesetzten 10s-Richtwert bestaetigt. Unter
// paralleler Last auf einem geteilten Modell-Host ist dieser Richtwert fuer
// p95/max nicht zu halten, ohne die Architektur umzubauen (ein einzelner
// Ollama-Prozess verarbeitet Anfragen serialisiert/teilserialisiert) — wie
// bei workflow-kanban QA-07 wird der Zielwert hier begruendet fuer den
// Lastfall angepasst: p50 bleibt bei 10s (typische Anfrage bleibt praktikabel),
// p95 wird auf 25s angehoben (Warteschlangenanteil unter Last explizit
// eingepreist statt verschwiegen).
func TestAsk_Real_ParallelLoad(t *testing.T) {
base := realBaseURL(t)
model := realModel(t)
p := ollama.New(base)
hits := []semanticsearch.Hit{
{SourceType: "document", SourceID: "vertrag-42", ChunkIndex: 0,
ChunkText: "Die Kuendigungsfrist fuer diesen Vertrag betraegt drei Monate zum Quartalsende.", Score: 0.95},
}
svc := New(stubSearcher{hits: hits}, p, model, nil)
const concurrency = 3
const sessions = 6
result := loadtest.Run(context.Background(), "docchat-ask-real", concurrency, sessions, func(ctx context.Context) error {
_, err := svc.Ask(ctx, "acme", "Wie lang ist die Kuendigungsfrist?", 5,
[]semanticsearch.Ref{{SourceType: "document", SourceID: "vertrag-42"}})
return err
})
t.Logf("QA-08 Chat-Last (real, gpt-oss:20b): concurrency=%d sessions=%d errors=%d dauer=%s durchsatz=%.2f/s p50=%s p95=%s p99=%s max=%s",
result.Concurrency, result.Sessions, result.Errors, result.Duration,
result.Throughput(), result.Percentile(0.5), result.Percentile(0.95), result.Percentile(0.99), result.Max())
if result.Errors > 0 {
t.Fatalf("%d von %d Chat-Anfragen fehlgeschlagen", result.Errors, result.Sessions)
}
if p50 := result.Percentile(0.5); p50 > 10*time.Second {
t.Fatalf("p50 = %s, ueberschreitet den angepassten Zielwert von 10s", p50)
}
if p95 := result.Percentile(0.95); p95 > 25*time.Second {
t.Fatalf("p95 = %s, ueberschreitet den angepassten Zielwert von 25s", p95)
}
}
// QA-08 Pruefung 1 (AK1), Mechanismus bei hoeherer Parallelitaet: der reale
// Ollama-Host auf office.sysops.de ist geteilte Infrastruktur, hoehere
// Nebenlaeufigkeit dort waere ein Lasttest gegen fremde Ressourcen statt
// gegen den eigenen Code. Mit einem Stub-Provider, dessen simulierte Latenz
// aus TestAsk_Real_ParallelLoad uebernommen wird (realistische Baseline statt
// erfundener Wert), wird geprueft, dass der eigene Anfrage-Pfad
// (Retrieval + Prompt-Aufbau + Zitate) bei zehnfacher Parallelitaet nicht
// selbst zum Flaschenhals wird.
func TestAsk_StubParallelLoad_HigherConcurrency(t *testing.T) {
hits := []semanticsearch.Hit{
{SourceType: "document", SourceID: "vertrag-42", ChunkIndex: 0,
ChunkText: "Die Kuendigungsfrist fuer diesen Vertrag betraegt drei Monate zum Quartalsende.", Score: 0.95},
}
simulatedLatency := 9200 * time.Millisecond // Baseline aus SEM-04 Einzeltest
if v := os.Getenv("QA08_SIMULATED_CHAT_LATENCY_MS"); v != "" {
if ms, err := time.ParseDuration(v + "ms"); err == nil {
simulatedLatency = ms
}
}
svc := New(stubSearcher{hits: hits}, &delayedProvider{delay: simulatedLatency}, "test-model", nil)
const concurrency = 30
const sessions = 90
result := loadtest.Run(context.Background(), "docchat-ask-stub", concurrency, sessions, func(ctx context.Context) error {
_, err := svc.Ask(ctx, "acme", "Wie lang ist die Kuendigungsfrist?", 5,
[]semanticsearch.Ref{{SourceType: "document", SourceID: "vertrag-42"}})
return err
})
t.Logf("QA-08 Chat-Last (Stub, Baseline-Latenz=%s): concurrency=%d sessions=%d errors=%d dauer=%s durchsatz=%.2f/s p50=%s p95=%s p99=%s max=%s",
simulatedLatency, result.Concurrency, result.Sessions, result.Errors, result.Duration,
result.Throughput(), result.Percentile(0.5), result.Percentile(0.95), result.Percentile(0.99), result.Max())
if result.Errors > 0 {
t.Fatalf("%d von %d Chat-Anfragen fehlgeschlagen", result.Errors, result.Sessions)
}
// Eigener Anfrage-Pfad darf ueber der simulierten Modell-Latenz hinaus
// keine spuerbare Zusatzverzoegerung einfuehren (AK1): Aufschlag max. 2s.
overhead := result.Max() - simulatedLatency
if overhead > 2*time.Second {
t.Fatalf("eigener Anfrage-Pfad fuegt %s Zusatzverzoegerung hinzu (max=%s, baseline=%s), erwartet <= 2s", overhead, result.Max(), simulatedLatency)
}
}
type delayedProvider struct {
delay time.Duration
calls atomic.Int64
}
func (p *delayedProvider) Chat(ctx context.Context, req provider.ChatRequest) (provider.ChatResponse, error) {
p.calls.Add(1)
select {
case <-time.After(p.delay):
case <-ctx.Done():
return provider.ChatResponse{}, ctx.Err()
}
return provider.ChatResponse{Model: req.Model, Message: provider.Message{Role: provider.RoleAssistant, Content: "Antwort laut Beleg [1]."}}, nil
}
func (p *delayedProvider) Complete(ctx context.Context, req provider.CompletionRequest) (provider.CompletionResponse, error) {
return provider.CompletionResponse{}, nil
}
func (p *delayedProvider) Embed(ctx context.Context, req provider.EmbeddingRequest) (provider.EmbeddingResponse, error) {
return provider.EmbeddingResponse{}, nil
}
+13 -3
View File
@@ -20,17 +20,22 @@ var ErrProcessingTimeout = errors.New("limits: verarbeitungszeit ueberschritten"
const (
DefaultMaxInputBytes = 200_000
DefaultMaxProcessingTime = 60 * time.Second
// DefaultMaxPendingJobs ist QA-08s Ueberlast-Grenzwert: oberhalb dieser
// Anzahl wartender Jobs lehnt Enqueue neue Arbeit ab, statt die
// Warteschlange unbegrenzt anwachsen zu lassen (AK3).
DefaultMaxPendingJobs = 500
)
type Limits struct {
MaxInputBytes int
MaxProcessingTime time.Duration
MaxInputBytes int
MaxProcessingTime time.Duration
MaxPendingJobs int
}
// FromEnv liest Obergrenzen aus Umgebungsvariablen, mit Standardwerten bei
// fehlender oder ungueltiger Angabe. 0 deaktiviert die jeweilige Grenze.
func FromEnv() Limits {
l := Limits{MaxInputBytes: DefaultMaxInputBytes, MaxProcessingTime: DefaultMaxProcessingTime}
l := Limits{MaxInputBytes: DefaultMaxInputBytes, MaxProcessingTime: DefaultMaxProcessingTime, MaxPendingJobs: DefaultMaxPendingJobs}
if v, ok := os.LookupEnv("AI_INFERENCE_MAX_INPUT_BYTES"); ok {
if n, err := strconv.Atoi(v); err == nil && n >= 0 {
l.MaxInputBytes = n
@@ -41,6 +46,11 @@ func FromEnv() Limits {
l.MaxProcessingTime = time.Duration(n) * time.Second
}
}
if v, ok := os.LookupEnv("AI_INFERENCE_MAX_PENDING_JOBS"); ok {
if n, err := strconv.Atoi(v); err == nil && n >= 0 {
l.MaxPendingJobs = n
}
}
return l
}
+138
View File
@@ -0,0 +1,138 @@
// Package loadtest implementiert die fuer QA-07/QA-08 gemeinsam genutzte
// Last- und Leistungsmessung: parallele Sessions ausfuehren, Latenzen
// aufzeichnen, Durchsatz bestimmen. k6/vegeta sind auf diesem Rechner nicht
// installierbar (Projektregel: keine zusaetzlichen Toolchains/Dienste
// installieren) — echter Ersatz ohne externe Abhaengigkeit: reale
// nebenlaeufige Aufrufe ueber die Go-Standardbibliothek, gegen den echten,
// in dieser Sitzung gebauten Code. Eigenstaendige Kopie aus
// mail/internal/loadtest, da Go-Module keine internal/-Pakete teilen.
package loadtest
import (
"context"
"os"
"runtime"
"sort"
"strconv"
"strings"
"sync"
"time"
)
// Result fasst Latenz- und Durchsatzwerte eines Lasttest-Laufs zusammen.
type Result struct {
Protocol string
Concurrency int
Sessions int
Errors int
Duration time.Duration
Latencies []time.Duration // aufsteigend sortiert
}
// Throughput liefert Sessions pro Sekunde über die Gesamtlaufzeit.
func (r Result) Throughput() float64 {
if r.Duration <= 0 {
return 0
}
return float64(r.Sessions) / r.Duration.Seconds()
}
// Percentile liefert die Latenz am p-ten Perzentil (0.0–1.0).
func (r Result) Percentile(p float64) time.Duration {
if len(r.Latencies) == 0 {
return 0
}
idx := int(p * float64(len(r.Latencies)-1))
if idx < 0 {
idx = 0
}
if idx >= len(r.Latencies) {
idx = len(r.Latencies) - 1
}
return r.Latencies[idx]
}
// Max liefert die höchste gemessene Latenz.
func (r Result) Max() time.Duration {
if len(r.Latencies) == 0 {
return 0
}
return r.Latencies[len(r.Latencies)-1]
}
// Run führt totalSessions Sessions über maximal concurrency parallele
// Worker aus und misst je Session die Latenz von sessionFunc.
func Run(ctx context.Context, protocol string, concurrency, totalSessions int, sessionFunc func(ctx context.Context) error) Result {
var mu sync.Mutex
latencies := make([]time.Duration, 0, totalSessions)
errCount := 0
sem := make(chan struct{}, concurrency)
var wg sync.WaitGroup
start := time.Now()
for i := 0; i < totalSessions; i++ {
sem <- struct{}{}
wg.Add(1)
go func() {
defer wg.Done()
defer func() { <-sem }()
t0 := time.Now()
err := sessionFunc(ctx)
elapsed := time.Since(t0)
mu.Lock()
latencies = append(latencies, elapsed)
if err != nil {
errCount++
}
mu.Unlock()
}()
}
wg.Wait()
duration := time.Since(start)
sort.Slice(latencies, func(i, j int) bool { return latencies[i] < latencies[j] })
return Result{
Protocol: protocol,
Concurrency: concurrency,
Sessions: totalSessions,
Errors: errCount,
Duration: duration,
Latencies: latencies,
}
}
// ResourceSample ist eine Momentaufnahme des Ressourcenverbrauchs DIESES
// Prozesses: Heap über die Go-Runtime, kumulierte CPU-Zeit über
// /proc/self/stat (Linux, kein externes Werkzeug nötig).
type ResourceSample struct {
HeapAllocBytes uint64
CPUTimeSeconds float64
}
func SampleResources() ResourceSample {
var m runtime.MemStats
runtime.ReadMemStats(&m)
return ResourceSample{HeapAllocBytes: m.HeapAlloc, CPUTimeSeconds: readSelfCPUTimeSeconds()}
}
// readSelfCPUTimeSeconds liefert die kumulierte User+System-CPU-Zeit dieses
// Prozesses in Sekunden. Liefert 0, wenn /proc/self/stat nicht lesbar ist
// (z. B. auf Nicht-Linux-Systemen) — Aufrufer behandeln das als "keine
// CPU-Messung verfügbar", kein harter Fehler.
func readSelfCPUTimeSeconds() float64 {
data, err := os.ReadFile("/proc/self/stat")
if err != nil {
return 0
}
fields := strings.Fields(string(data))
if len(fields) < 15 {
return 0
}
utime, err1 := strconv.ParseFloat(fields[13], 64)
stime, err2 := strconv.ParseFloat(fields[14], 64)
if err1 != nil || err2 != nil {
return 0
}
const clockTicksPerSecond = 100 // Linux-Standard-USER_HZ
return (utime + stime) / clockTicksPerSecond
}
+14
View File
@@ -29,6 +29,11 @@ const (
var ErrNotFound = errors.New("queue: job nicht gefunden")
var ErrNoJobAvailable = errors.New("queue: kein job verfuegbar")
// ErrQueueFull ist QA-08s definiertes Ueberlastverhalten (AK3): Enqueue
// lehnt neue Jobs kontrolliert ab, statt die Warteschlange unbegrenzt
// anwachsen zu lassen, wenn bereits limits.MaxPendingJobs Jobs warten.
var ErrQueueFull = errors.New("queue: warteschlange ueberlastet, bitte spaeter erneut versuchen")
// Kind unterscheidet die Art des Inferenz-Aufrufs im Payload.
type Kind string
@@ -104,6 +109,15 @@ func (q *Queue) Enqueue(ctx context.Context, jobType Kind, payload any, opts Enq
return "", err
}
}
if q.limits.MaxPendingJobs > 0 {
depth, err := q.Depth(ctx)
if err != nil {
return "", err
}
if depth.Pending >= int64(q.limits.MaxPendingJobs) {
return "", ErrQueueFull
}
}
maxAttempts := opts.MaxAttempts
if maxAttempts <= 0 {
maxAttempts = DefaultMaxAttempts
+116
View File
@@ -0,0 +1,116 @@
package queue
import (
"context"
"errors"
"sync"
"testing"
"time"
"gitea.perlbach24.de/scripte/nexarch/ai/internal/limits"
"gitea.perlbach24.de/scripte/nexarch/ai/internal/loadtest"
"gitea.perlbach24.de/scripte/nexarch/ai/internal/provider"
)
// QA-08 Pruefung 2 (AK2): realistische parallele Klassifizierungs-/
// Extraktionsanfragen fuellen die Queue, mehrere Worker laufen gleichzeitig
// dagegen — die Warteschlangentiefe waechst waehrend des Laufs nicht
// unkontrolliert, weil Dequeue/Complete nebenlaeufig abraeumen.
func TestQueue_DepthStaysBoundedUnderParallelLoad(t *testing.T) {
pool := newPool(t)
q := NewQueue(pool, time.Minute)
q.SetLimits(limits.Limits{MaxPendingJobs: 0}) // in diesem Lauf keine Enqueue-Obergrenze, es geht um die Abarbeitung
const workers = 4
const totalJobs = 120
stop := make(chan struct{})
var wg sync.WaitGroup
for i := 0; i < workers; i++ {
wg.Add(1)
w := NewWorker(q, &fakeProvider{}, "loadtest-worker")
go func() {
defer wg.Done()
for {
select {
case <-stop:
return
default:
}
if err := w.ProcessOne(context.Background()); err != nil && !errors.Is(err, ErrNoJobAvailable) {
t.Logf("worker fehler (nicht fatal, da nebenlaeufig): %v", err)
}
}
}()
}
result := loadtest.Run(context.Background(), "inference-queue-enqueue", 20, totalJobs, func(ctx context.Context) error {
_, err := q.Enqueue(ctx, KindChat, provider.ChatRequest{Model: "test"}, EnqueueOptions{})
return err
})
deadline := time.Now().Add(20 * time.Second)
var depth Depth
for time.Now().Before(deadline) {
var err error
depth, err = q.Depth(context.Background())
if err != nil {
t.Fatalf("depth: %v", err)
}
if depth.Pending == 0 {
break
}
time.Sleep(100 * time.Millisecond)
}
close(stop)
wg.Wait()
t.Logf("QA-08 Enqueue-Last: concurrency=%d sessions=%d errors=%d dauer=%s durchsatz=%.1f/s p50=%s p95=%s p99=%s max=%s",
result.Concurrency, result.Sessions, result.Errors, result.Duration,
result.Throughput(), result.Percentile(0.5), result.Percentile(0.95), result.Percentile(0.99), result.Max())
t.Logf("QA-08 Warteschlangenstand nach Abarbeitung: pending=%d processing=%d dead_letter=%d", depth.Pending, depth.Processing, depth.DeadLetter)
if result.Errors > 0 {
t.Fatalf("%d von %d Enqueue-Aufrufen fehlgeschlagen, erwartet 0 ohne Ueberlast-Grenze", result.Errors, result.Sessions)
}
if depth.Pending != 0 {
t.Fatalf("depth.Pending = %d nach Ablauf der Wartezeit, erwartet 0 (Warteschlange sollte abgearbeitet sein, kein unkontrolliertes Wachstum)", depth.Pending)
}
}
// QA-08 Pruefung 3 (AK3): bei erreichter Ueberlast-Grenze ist das Verhalten
// definiert — Enqueue lehnt kontrolliert mit ErrQueueFull ab, statt
// unbegrenzt weiter Jobs anzunehmen.
func TestQueue_RejectsEnqueueWhenQueueFull(t *testing.T) {
pool := newPool(t)
q := NewQueue(pool, time.Minute)
q.SetLimits(limits.Limits{MaxPendingJobs: 5})
for i := 0; i < 5; i++ {
if _, err := q.Enqueue(context.Background(), KindChat, provider.ChatRequest{Model: "test"}, EnqueueOptions{}); err != nil {
t.Fatalf("enqueue %d (sollte noch innerhalb der Grenze liegen): %v", i, err)
}
}
_, err := q.Enqueue(context.Background(), KindChat, provider.ChatRequest{Model: "test"}, EnqueueOptions{})
if !errors.Is(err, ErrQueueFull) {
t.Fatalf("enqueue beim 6. Job = %v, erwartet ErrQueueFull", err)
}
depth, err := q.Depth(context.Background())
if err != nil {
t.Fatalf("depth: %v", err)
}
if depth.Pending != 5 {
t.Fatalf("depth.Pending = %d, erwartet 5 (abgelehnter Job wurde nicht angelegt)", depth.Pending)
}
// Nach Abarbeitung eines Jobs ist unter der Grenze wieder Platz (definiertes
// Verhalten, kein dauerhafter Deadlock der Warteschlange).
w := NewWorker(q, &fakeProvider{}, "worker-1")
if err := w.ProcessOne(context.Background()); err != nil {
t.Fatalf("processone: %v", err)
}
if _, err := q.Enqueue(context.Background(), KindChat, provider.ChatRequest{Model: "test"}, EnqueueOptions{}); err != nil {
t.Fatalf("enqueue nach freigewordenem platz: %v", err)
}
}