nexarch/dms/internal/jobqueue/worker_limits_test.go

package jobqueue

import (
	"context"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"
)

type fakeNotifier struct {
	mu    sync.Mutex
	calls []string
}

func (f *fakeNotifier) NotifyJobAborted(_ context.Context, job *Job, reason string) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.calls = append(f.calls, job.ID+": "+reason)
	return nil
}

func (f *fakeNotifier) count() int {
	f.mu.Lock()
	defer f.mu.Unlock()
	return len(f.calls)
}

// TestWorker_InputLimitExceeded_AbortsWithoutRunningHandler ist Akzeptanz-
// kriterium 1+2 sowie Pruefung 1 aus dem Ticket: eine Datei, die die
// Seitenobergrenze ueberschreitet, fuehrt zum kontrollierten Abbruch
// (Dead-Letter-Queue), OHNE dass der Handler ueberhaupt gestartet wird.
// Zugleich Pruefung 3 (Nutzerbenachrichtigung).
func TestWorker_InputLimitExceeded_AbortsWithoutRunningHandler(t *testing.T) {
	pool := setupTest(t)
	q := NewQueue(pool, time.Minute)
	ctx := context.Background()

	id, err := q.Enqueue(ctx, "convert", map[string]any{"pages": 5000}, EnqueueOptions{})
	if err != nil {
		t.Fatalf("enqueue: %v", err)
	}

	var handlerCalled int32
	notifier := &fakeNotifier{}
	w := NewWorker(q, "worker-1", nil, time.Millisecond, func(ctx context.Context, job *Job) error {
		atomic.AddInt32(&handlerCalled, 1)
		return nil
	}, WithDefaultLimits(Limits{MaxDuration: time.Second, MaxPages: 100}), WithNotifier(notifier))

	w.RunOnce(ctx)

	if atomic.LoadInt32(&handlerCalled) != 0 {
		t.Fatalf("handler wurde trotz Seitenlimit-Ueberschreitung aufgerufen")
	}
	status, err := q.Status(ctx, id)
	if err != nil {
		t.Fatalf("status: %v", err)
	}
	if status.Status != StatusDeadLetter {
		t.Fatalf("status = %q, want %q", status.Status, StatusDeadLetter)
	}
	if status.LastError == nil || !strings.Contains(*status.LastError, "seitenzahl") {
		t.Fatalf("last_error = %v, want hinweis auf seitenzahl", status.LastError)
	}
	if notifier.count() != 1 {
		t.Fatalf("notifier.calls = %d, want 1 (nutzer muss informiert werden)", notifier.count())
	}
}

// TestWorker_HandlerExceedsMaxDuration_AbortsControlled ist Akzeptanz-
// kriterium 2: ein Handler, der die Zeitobergrenze ueberschreitet
// (und den Kontext ignoriert), wird kontrolliert abgebrochen. RunOnce kehrt
// dabei nahe der Obergrenze zurueck statt erst nach Handler-Ende — der
// Worker wird also nicht blockiert (Akzeptanzkriterium 3).
func TestWorker_HandlerExceedsMaxDuration_AbortsControlled(t *testing.T) {
	pool := setupTest(t)
	q := NewQueue(pool, time.Minute)
	ctx := context.Background()

	id, err := q.Enqueue(ctx, "ocr", nil, EnqueueOptions{})
	if err != nil {
		t.Fatalf("enqueue: %v", err)
	}

	notifier := &fakeNotifier{}
	w := NewWorker(q, "worker-1", nil, time.Millisecond, func(ctx context.Context, job *Job) error {
		// Ignoriert absichtlich ctx.Done(), simuliert einen entgleisten Job
		// (z.B. hängende Konvertierungsbibliothek).
		time.Sleep(2 * time.Second)
		return nil
	}, WithDefaultLimits(Limits{MaxDuration: 50 * time.Millisecond}), WithNotifier(notifier))

	start := time.Now()
	w.RunOnce(ctx)
	elapsed := time.Since(start)

	if elapsed > 500*time.Millisecond {
		t.Fatalf("RunOnce dauerte %s, Worker wurde durch entgleisten Job blockiert (wollte < 500ms)", elapsed)
	}
	status, err := q.Status(ctx, id)
	if err != nil {
		t.Fatalf("status: %v", err)
	}
	if status.Status != StatusDeadLetter {
		t.Fatalf("status = %q, want %q", status.Status, StatusDeadLetter)
	}
	if status.LastError == nil || !strings.Contains(*status.LastError, "verarbeitungszeit") {
		t.Fatalf("last_error = %v, want hinweis auf verarbeitungszeit", status.LastError)
	}
	if notifier.count() != 1 {
		t.Fatalf("notifier.calls = %d, want 1", notifier.count())
	}
}

// TestWorker_MemoryLimitExceeded_AbortsControlled ist Akzeptanzkriterium 1/2
// fuer die Speicherobergrenze: ein Handler, der kontinuierlich Speicher
// allokiert, wird abgebrochen bevor er die MaxDuration erreicht.
func TestWorker_MemoryLimitExceeded_AbortsControlled(t *testing.T) {
	pool := setupTest(t)
	q := NewQueue(pool, time.Minute)
	ctx := context.Background()

	id, err := q.Enqueue(ctx, "convert", nil, EnqueueOptions{})
	if err != nil {
		t.Fatalf("enqueue: %v", err)
	}

	notifier := &fakeNotifier{}
	w := NewWorker(q, "worker-1", nil, time.Millisecond, func(ctx context.Context, job *Job) error {
		buf := make([][]byte, 0)
		for i := 0; i < 500; i++ {
			select {
			case <-ctx.Done():
				return ctx.Err()
			default:
			}
			buf = append(buf, make([]byte, 2*1024*1024)) // 2 MiB je Schritt
			time.Sleep(5 * time.Millisecond)
		}
		return nil
	}, WithDefaultLimits(Limits{MaxDuration: 10 * time.Second, MaxMemoryBytes: 8 * 1024 * 1024}), WithNotifier(notifier))

	start := time.Now()
	w.RunOnce(ctx)
	elapsed := time.Since(start)

	if elapsed > 3*time.Second {
		t.Fatalf("RunOnce dauerte %s, wollte deutlich unter der MaxDuration abbrechen", elapsed)
	}
	status, err := q.Status(ctx, id)
	if err != nil {
		t.Fatalf("status: %v", err)
	}
	if status.Status != StatusDeadLetter {
		t.Fatalf("status = %q, want %q (last_error=%v)", status.Status, StatusDeadLetter, status.LastError)
	}
	if notifier.count() != 1 {
		t.Fatalf("notifier.calls = %d, want 1", notifier.count())
	}
}

// TestWorker_ExceedingJob_DoesNotBlockNextJobOnSameWorker ist Pruefung 2
// (in reduzierter, deterministischer Form): derselbe Worker verarbeitet nach
// einem kontrolliert abgebrochenen, entgleisten Job sofort den naechsten
// regulaeren Job weiter, statt auf das (im Hintergrund weiterlaufende)
// Handler-Ende des ersten Jobs zu warten.
func TestWorker_ExceedingJob_DoesNotBlockNextJobOnSameWorker(t *testing.T) {
	pool := setupTest(t)
	q := NewQueue(pool, time.Minute)
	ctx := context.Background()

	slowID, err := q.Enqueue(ctx, "slow-job", nil, EnqueueOptions{})
	if err != nil {
		t.Fatalf("enqueue slow: %v", err)
	}
	time.Sleep(5 * time.Millisecond) // available_at-Reihenfolge sichern
	normalID, err := q.Enqueue(ctx, "normal-job", nil, EnqueueOptions{})
	if err != nil {
		t.Fatalf("enqueue normal: %v", err)
	}

	var normalHandled int32
	w := NewWorker(q, "worker-1", nil, time.Millisecond, func(ctx context.Context, job *Job) error {
		if job.JobType == "slow-job" {
			time.Sleep(2 * time.Second) // ignoriert ctx absichtlich
			return nil
		}
		atomic.AddInt32(&normalHandled, 1)
		return nil
	},
		WithJobTypeLimits("slow-job", Limits{MaxDuration: 30 * time.Millisecond}),
		WithDefaultLimits(Limits{MaxDuration: time.Second}),
	)

	start := time.Now()
	w.RunOnce(ctx) // slow-job: bricht nach ~30ms kontrolliert ab
	w.RunOnce(ctx) // normal-job: muss sofort verarbeitet werden
	elapsed := time.Since(start)

	if elapsed > 1*time.Second {
		t.Fatalf("beide RunOnce-Aufrufe dauerten zusammen %s, normal-job wurde durch slow-job blockiert", elapsed)
	}
	if atomic.LoadInt32(&normalHandled) != 1 {
		t.Fatalf("normal-job wurde nicht verarbeitet")
	}
	slowStatus, err := q.Status(ctx, slowID)
	if err != nil {
		t.Fatalf("status slow: %v", err)
	}
	if slowStatus.Status != StatusDeadLetter {
		t.Fatalf("slow-job status = %q, want %q", slowStatus.Status, StatusDeadLetter)
	}
	normalStatus, err := q.Status(ctx, normalID)
	if err != nil {
		t.Fatalf("status normal: %v", err)
	}
	if normalStatus.Status != StatusSucceeded {
		t.Fatalf("normal-job status = %q, want %q", normalStatus.Status, StatusSucceeded)
	}
}

// TestWorker_ExceedingJobDoesNotBlockParallelRegularJobs ist der Lasttest
// aus Pruefung 2: mehrere Worker-Aufrufe (mehrere Worker-Prozesse simulierend,
// vgl. FDN-04-Design "mehrinstanzfaehig") laufen gleichzeitig gegen dieselbe
// Queue; ein ueberschreitender Job (Seitenlimit) blockiert die parallele
// Verarbeitung der regulaeren Jobs nicht.
func TestWorker_ExceedingJobDoesNotBlockParallelRegularJobs(t *testing.T) {
	pool := setupTest(t)
	q := NewQueue(pool, time.Minute)
	ctx := context.Background()

	if _, err := q.Enqueue(ctx, "convert", map[string]any{"pages": 99999}, EnqueueOptions{}); err != nil {
		t.Fatalf("enqueue oversized: %v", err)
	}
	const normalJobs = 10
	for i := 0; i < normalJobs; i++ {
		if _, err := q.Enqueue(ctx, "convert", map[string]any{"pages": 1}, EnqueueOptions{}); err != nil {
			t.Fatalf("enqueue normal %d: %v", i, err)
		}
	}

	var completed int32
	notifier := &fakeNotifier{}
	w := NewWorker(q, "worker-1", nil, time.Millisecond, func(ctx context.Context, job *Job) error {
		atomic.AddInt32(&completed, 1)
		return nil
	}, WithDefaultLimits(Limits{MaxDuration: time.Second, MaxPages: 100}), WithNotifier(notifier))

	start := time.Now()
	var wg sync.WaitGroup
	for i := 0; i < normalJobs+1; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			w.RunOnce(ctx)
		}()
	}
	wg.Wait()
	elapsed := time.Since(start)

	if elapsed > 2*time.Second {
		t.Fatalf("parallele verarbeitung dauerte %s, uebergrosser job hat andere jobs blockiert", elapsed)
	}
	if atomic.LoadInt32(&completed) != normalJobs {
		t.Fatalf("completed = %d, want %d", completed, normalJobs)
	}
	if notifier.count() != 1 {
		t.Fatalf("notifier.calls = %d, want 1 (nur der uebergrosse job)", notifier.count())
	}
}