package queue import ( "context" "errors" "sync" "testing" "time" "gitea.perlbach24.de/scripte/nexarch/ai/internal/limits" "gitea.perlbach24.de/scripte/nexarch/ai/internal/provider" "gitea.perlbach24.de/scripte/nexarch/common/loadtest" ) // 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) } }