nexarch/ai/internal/queue/queue_loadtest_test.go

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)
	}
}