nexarch/ai/internal/queue/queue_test.go

package queue

import (
	"context"
	"errors"
	"os"
	"path/filepath"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"

	"gitea.perlbach24.de/scripte/nexarch/ai/internal/migrate"
	"gitea.perlbach24.de/scripte/nexarch/ai/internal/provider"
	"gitea.perlbach24.de/scripte/nexarch/common/testdb"
)

func newPool(t *testing.T) *pgxpool.Pool {
	t.Helper()
	dsn := os.Getenv("TEST_TENANT_DSN")
	if dsn == "" {
		t.Skip("TEST_TENANT_DSN nicht gesetzt, Integrationstest uebersprungen")
	}
	pool := testdb.Pool(t, dsn)

	root, err := filepath.Abs("../../migrations")
	if err != nil {
		t.Fatalf("migrationsverzeichnis: %v", err)
	}
	migrations, err := migrate.Load(root)
	if err != nil {
		t.Fatalf("migrationen laden: %v", err)
	}
	if _, err := migrate.Up(context.Background(), pool, migrations); err != nil {
		t.Fatalf("migrationen anwenden: %v", err)
	}
	return pool
}

// AK1 + Pruefung 1: Inferenz-Aufrufe laufen asynchron, eine Lastspitze
// (viele gleichzeitige Enqueue-Aufrufe von mehreren nebenlaeufigen
// "Anfrage"-Goroutinen) blockiert nicht — jeder Enqueue-Aufruf kehrt sofort
// zurueck, unabhaengig davon ob der Worker gerade etwas verarbeitet.
func TestQueue_AsyncEnqueueUnderLoadSpike(t *testing.T) {
	pool := newPool(t)
	q := NewQueue(pool, time.Minute)

	const n = 50
	var wg sync.WaitGroup
	ids := make([]string, n)
	start := time.Now()
	for i := 0; i < n; i++ {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			id, err := q.Enqueue(context.Background(), KindChat, provider.ChatRequest{Model: "test"}, EnqueueOptions{})
			if err != nil {
				t.Errorf("enqueue %d: %v", i, err)
				return
			}
			ids[i] = id
		}(i)
	}
	wg.Wait()
	elapsed := time.Since(start)
	if elapsed > 5*time.Second {
		t.Fatalf("enqueue von %d anfragen dauerte %s, das deutet auf Blockierung hin", n, elapsed)
	}
	for i, id := range ids {
		if id == "" {
			t.Fatalf("job %d hat keine ID erhalten", i)
		}
	}
	depth, err := q.Depth(context.Background())
	if err != nil {
		t.Fatalf("depth: %v", err)
	}
	if depth.Pending < n {
		t.Fatalf("depth.Pending = %d, erwartet mindestens %d", depth.Pending, n)
	}
}

type fakeProvider struct {
	calls   atomic.Int64
	failN   int64 // erste failN Aufrufe schlagen fehl
	failure error
}

func (f *fakeProvider) Chat(ctx context.Context, req provider.ChatRequest) (provider.ChatResponse, error) {
	n := f.calls.Add(1)
	if n <= f.failN {
		return provider.ChatResponse{}, f.failure
	}
	return provider.ChatResponse{Model: req.Model, Message: provider.Message{Role: provider.RoleAssistant, Content: "ok"}}, nil
}
func (f *fakeProvider) Complete(ctx context.Context, req provider.CompletionRequest) (provider.CompletionResponse, error) {
	return provider.CompletionResponse{}, nil
}
func (f *fakeProvider) Embed(ctx context.Context, req provider.EmbeddingRequest) (provider.EmbeddingResponse, error) {
	return provider.EmbeddingResponse{}, nil
}

// AK2 + Pruefung 2: ein fehlgeschlagener Aufruf wird mit Rueckverfolgung
// (Versuchszaehler, letzter Fehler) wiederholt, bis er entweder gelingt oder
// die maximale Versuchszahl erreicht ist (dann dead_letter).
func TestQueue_RetryAfterFailure(t *testing.T) {
	pool := newPool(t)
	q := NewQueue(pool, time.Minute)
	fp := &fakeProvider{failN: 2, failure: &provider.Error{Provider: "fake", Code: provider.CodeUpstream, Message: "boom"}}
	w := NewWorker(q, fp, "worker-1")

	id, err := q.Enqueue(context.Background(), KindChat, provider.ChatRequest{Model: "test"}, EnqueueOptions{MaxAttempts: 5})
	if err != nil {
		t.Fatalf("enqueue: %v", err)
	}

	// Versuch 1: schlaegt fehl.
	if err := w.ProcessOne(context.Background()); err == nil {
		t.Fatal("erster Versuch haette fehlschlagen sollen")
	}
	st, err := q.Status(context.Background(), id)
	if err != nil {
		t.Fatalf("status: %v", err)
	}
	if st.Status != StatusPending {
		t.Fatalf("status nach 1. Fehlschlag = %q, erwartet pending (fuer erneuten Versuch)", st.Status)
	}
	if st.Attempts != 1 {
		t.Fatalf("attempts = %d, erwartet 1 (Rueckverfolgung)", st.Attempts)
	}
	if st.LastError == nil || *st.LastError == "" {
		t.Fatal("last_error sollte den letzten Fehler festhalten")
	}

	// Backoff nach vorne verlegen, damit der naechste Dequeue den Job sofort holt.
	if _, err := pool.Exec(context.Background(), `UPDATE inference_jobs SET available_at = now() WHERE id = $1`, id); err != nil {
		t.Fatalf("available_at vorziehen: %v", err)
	}

	// Versuch 2: schlaegt erneut fehl (failN=2).
	if err := w.ProcessOne(context.Background()); err == nil {
		t.Fatal("zweiter Versuch haette fehlschlagen sollen")
	}
	if _, err := pool.Exec(context.Background(), `UPDATE inference_jobs SET available_at = now() WHERE id = $1`, id); err != nil {
		t.Fatalf("available_at vorziehen: %v", err)
	}

	// Versuch 3: gelingt (failN=2, dritter Aufruf laesst durch).
	if err := w.ProcessOne(context.Background()); err != nil {
		t.Fatalf("dritter Versuch haette gelingen sollen: %v", err)
	}
	st, err = q.Status(context.Background(), id)
	if err != nil {
		t.Fatalf("status: %v", err)
	}
	if st.Status != StatusSucceeded {
		t.Fatalf("status nach Erfolg = %q, erwartet succeeded", st.Status)
	}
	if st.Attempts != 3 {
		t.Fatalf("attempts = %d, erwartet 3", st.Attempts)
	}
}

// Erschoepfte Versuche wandern in die Dead-Letter-Queue statt endlos zu wiederholen.
func TestQueue_ExhaustedRetriesGoToDeadLetter(t *testing.T) {
	pool := newPool(t)
	q := NewQueue(pool, time.Minute)
	fp := &fakeProvider{failN: 1000, failure: errors.New("dauerhaft kaputt")}
	w := NewWorker(q, fp, "worker-1")

	id, err := q.Enqueue(context.Background(), KindChat, provider.ChatRequest{Model: "test"}, EnqueueOptions{MaxAttempts: 2})
	if err != nil {
		t.Fatalf("enqueue: %v", err)
	}
	for i := 0; i < 2; i++ {
		_ = w.ProcessOne(context.Background())
		if _, err := pool.Exec(context.Background(), `UPDATE inference_jobs SET available_at = now() WHERE id = $1`, id); err != nil {
			t.Fatalf("available_at vorziehen: %v", err)
		}
	}
	st, err := q.Status(context.Background(), id)
	if err != nil {
		t.Fatalf("status: %v", err)
	}
	if st.Status != StatusDeadLetter {
		t.Fatalf("status nach erschoepften Versuchen = %q, erwartet dead_letter", st.Status)
	}
}

// AK3 + Pruefung 3: der Warteschlangenstand ist beobachtbar (fuer Monitoring).
func TestQueue_DepthIsObservable(t *testing.T) {
	pool := newPool(t)
	q := NewQueue(pool, time.Minute)

	before, err := q.Depth(context.Background())
	if err != nil {
		t.Fatalf("depth: %v", err)
	}
	if _, err := q.Enqueue(context.Background(), KindEmbedding, provider.EmbeddingRequest{Model: "test"}, EnqueueOptions{}); err != nil {
		t.Fatalf("enqueue: %v", err)
	}
	after, err := q.Depth(context.Background())
	if err != nil {
		t.Fatalf("depth: %v", err)
	}
	if after.Pending != before.Pending+1 {
		t.Fatalf("depth.Pending = %d, erwartet %d", after.Pending, before.Pending+1)
	}
}

// Dequeue-Semantik: Keine doppelte Zustellung an mehrere gleichzeitig
// pollende Worker (FOR UPDATE SKIP LOCKED).
func TestQueue_DequeueIsExclusive(t *testing.T) {
	pool := newPool(t)
	q := NewQueue(pool, time.Minute)
	id, err := q.Enqueue(context.Background(), KindChat, provider.ChatRequest{Model: "test"}, EnqueueOptions{})
	if err != nil {
		t.Fatalf("enqueue: %v", err)
	}

	var wg sync.WaitGroup
	var got atomic.Int64
	for i := 0; i < 5; i++ {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			j, err := q.Dequeue(context.Background(), "worker")
			if err == nil && j.ID == id {
				got.Add(1)
			}
		}(i)
	}
	wg.Wait()
	if got.Load() != 1 {
		t.Fatalf("job wurde %d mal zugestellt, erwartet genau 1", got.Load())
	}
}