nexarch/ai/internal/queue/worker_test.go

package queue

import (
	"context"
	"errors"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"gitea.perlbach24.de/scripte/nexarch/ai/internal/limits"
	"gitea.perlbach24.de/scripte/nexarch/ai/internal/provider"
)

// slowProvider blockiert bis ctx endet oder delay verstrichen ist, damit sich
// eine Zeitueberschreitung real simulieren laesst (respektiert ctx wie ein
// echter HTTP-Client es tun wuerde).
type slowProvider struct {
	delay      time.Duration
	calls      atomic.Int64
	cancelled  atomic.Int64
}

func (s *slowProvider) Chat(ctx context.Context, req provider.ChatRequest) (provider.ChatResponse, error) {
	s.calls.Add(1)
	select {
	case <-time.After(s.delay):
		return provider.ChatResponse{Model: req.Model}, nil
	case <-ctx.Done():
		s.cancelled.Add(1)
		return provider.ChatResponse{}, ctx.Err()
	}
}
func (s *slowProvider) Complete(ctx context.Context, req provider.CompletionRequest) (provider.CompletionResponse, error) {
	return provider.CompletionResponse{}, nil
}
func (s *slowProvider) Embed(ctx context.Context, req provider.EmbeddingRequest) (provider.EmbeddingResponse, error) {
	return provider.EmbeddingResponse{}, nil
}

// AK1 + Pruefung 3 (Regression): eine Anfrage innerhalb der Grenzen laeuft
// wie bisher durch.
func TestQueue_EnqueueWithinLimitsSucceeds(t *testing.T) {
	q := newQueueWithLimits(t, limits.Limits{MaxInputBytes: 1000, MaxProcessingTime: time.Second})
	id, err := q.Enqueue(context.Background(), KindChat, provider.ChatRequest{Model: "test", Messages: []provider.Message{{Content: "kurz"}}}, EnqueueOptions{})
	if err != nil {
		t.Fatalf("erwartet: kein fehler, erhalten: %v", err)
	}
	if id == "" {
		t.Fatal("erwartet: job-id")
	}
}

// AK1 + AK2 + Pruefung 1: eine Eingabe, die die Groessenobergrenze
// ueberschreitet, wird kontrolliert abgelehnt, statt in die Warteschlange
// aufgenommen zu werden.
func TestQueue_EnqueueExceedingSizeLimitIsRejected(t *testing.T) {
	q := newQueueWithLimits(t, limits.Limits{MaxInputBytes: 20, MaxProcessingTime: time.Second})
	big := provider.ChatRequest{Model: "test", Messages: []provider.Message{{Content: "dies ist eine viel zu lange nachricht fuer das limit"}}}
	_, err := q.Enqueue(context.Background(), KindChat, big, EnqueueOptions{})
	if !errors.Is(err, limits.ErrInputTooLarge) {
		t.Fatalf("erwartet: %v, erhalten: %v", limits.ErrInputTooLarge, err)
	}
}

// AK2 + AK3 + Pruefung 2: eine Anfrage, die die maximale Verarbeitungszeit
// ueberschreitet, wird kontrolliert abgelehnt (Fail statt Absturz/Haenger),
// und eine parallel laufende, normale Anfrage eines anderen "Mandanten"
// bleibt unbeeinflusst und schliesst regulaer ab.
func TestQueue_ProcessingTimeoutDoesNotBlockOtherJobs(t *testing.T) {
	pool := newPool(t)
	q := NewQueue(pool, 30*time.Second)
	q.SetLimits(limits.Limits{MaxInputBytes: 0, MaxProcessingTime: 50 * time.Millisecond})

	slow := &slowProvider{delay: 2 * time.Second}
	slowWorker := NewWorker(q, slow, "worker-slow")
	slowWorker.SetLimits(limits.Limits{MaxInputBytes: 0, MaxProcessingTime: 50 * time.Millisecond})

	fast := &fakeProvider{}
	fastWorker := NewWorker(q, fast, "worker-fast")
	fastWorker.SetLimits(limits.Limits{MaxInputBytes: 0, MaxProcessingTime: 5 * time.Second})

	slowID, err := q.Enqueue(context.Background(), KindChat, provider.ChatRequest{Model: "slow"}, EnqueueOptions{})
	if err != nil {
		t.Fatalf("enqueue slow: %v", err)
	}
	fastID, err := q.Enqueue(context.Background(), KindChat, provider.ChatRequest{Model: "fast"}, EnqueueOptions{})
	if err != nil {
		t.Fatalf("enqueue fast: %v", err)
	}

	var wg sync.WaitGroup
	wg.Add(2)
	start := time.Now()
	go func() {
		defer wg.Done()
		_ = slowWorker.ProcessOne(context.Background())
	}()
	go func() {
		defer wg.Done()
		if err := fastWorker.ProcessOne(context.Background()); err != nil {
			t.Errorf("fast job sollte trotz paralleler ueberschreitender anfrage durchlaufen: %v", err)
		}
	}()
	wg.Wait()
	elapsed := time.Since(start)

	if elapsed > time.Second {
		t.Fatalf("fast job wurde von der langsamen anfrage blockiert, dauer: %v", elapsed)
	}

	// Welcher Job von welchem Worker gezogen wurde, ist wegen FOR UPDATE SKIP
	// LOCKED nicht deterministisch (AK3: beide Worker laufen unabhaengig).
	// Entscheidend ist nur: genau ein Job zeigt die kontrollierte
	// Zeitueberschreitung, der andere ist unbeeinflusst erfolgreich.
	jobA, err := q.Status(context.Background(), slowID)
	if err != nil {
		t.Fatalf("status a: %v", err)
	}
	jobB, err := q.Status(context.Background(), fastID)
	if err != nil {
		t.Fatalf("status b: %v", err)
	}

	var timedOut, succeeded *Job
	for _, j := range []*Job{jobA, jobB} {
		switch j.Status {
		case StatusSucceeded:
			succeeded = j
		case StatusPending, StatusDeadLetter:
			timedOut = j
		}
	}
	if timedOut == nil || succeeded == nil {
		t.Fatalf("erwartet: genau ein timeout und ein erfolg, erhalten: a=%s b=%s", jobA.Status, jobB.Status)
	}
	if timedOut.LastError == nil || !containsTimeoutMessage(*timedOut.LastError) {
		t.Fatalf("erwartet: fehlermeldung zur zeitueberschreitung, erhalten: %v", timedOut.LastError)
	}
}

func containsTimeoutMessage(s string) bool {
	return len(s) > 0 && (contains(s, "verarbeitungszeit") || contains(s, "context deadline"))
}

func contains(s, substr string) bool {
	for i := 0; i+len(substr) <= len(s); i++ {
		if s[i:i+len(substr)] == substr {
			return true
		}
	}
	return false
}

func newQueueWithLimits(t *testing.T, l limits.Limits) *Queue {
	t.Helper()
	pool := newPool(t)
	q := NewQueue(pool, 30*time.Second)
	q.SetLimits(l)
	return q
}