nexarch/dms/internal/jobqueue/worker.go

package jobqueue

import (
	"context"
	"errors"
	"fmt"
	"log"
	"runtime"
	"time"

	"gitea.perlbach24.de/scripte/nexarch/dms/internal/logging"
)

// Handler verarbeitet einen einzelnen Job. Ein zurueckgegebener Fehler
// fuehrt zu Queue.Fail (Backoff/DLQ), nil zu Queue.Complete.
type Handler func(ctx context.Context, job *Job) error

// memoryPollInterval steuert, wie oft waehrend eines Handler-Laufs der
// Speicherverbrauch abgetastet wird (FDN-11).
const memoryPollInterval = 20 * time.Millisecond

// Worker pollt die Queue in einer In-Prozess-Goroutine und ruft Handler je
// abgeholtem Job auf — die "In-Prozess-Worker-Goroutinen" aus der
// Ticket-Vorgabe, kein separater Prozess/Redis noetig.
type Worker struct {
	queue         *Queue
	workerID      string
	jobTypes      []string
	pollInterval  time.Duration
	handler       Handler
	defaultLimits Limits
	jobTypeLimits map[string]Limits
	notifier      Notifier
}

// WorkerOption konfiguriert optionale FDN-11-Ressourcenlimits/Benachrichtigung.
type WorkerOption func(*Worker)

// WithDefaultLimits ersetzt die Vorgabe-Obergrenzen (DefaultLimits) fuer
// Jobtypen ohne eigene WithJobTypeLimits-Konfiguration.
func WithDefaultLimits(l Limits) WorkerOption {
	return func(w *Worker) { w.defaultLimits = l }
}

// WithJobTypeLimits setzt Obergrenzen fuer einen einzelnen Jobtyp
// (Akzeptanzkriterium 1: konfigurierbar je Job).
func WithJobTypeLimits(jobType string, l Limits) WorkerOption {
	return func(w *Worker) {
		if w.jobTypeLimits == nil {
			w.jobTypeLimits = map[string]Limits{}
		}
		w.jobTypeLimits[jobType] = l
	}
}

// WithNotifier ersetzt den Standard-LogNotifier durch eine andere
// Benachrichtigungs-Implementierung.
func WithNotifier(n Notifier) WorkerOption {
	return func(w *Worker) { w.notifier = n }
}

func NewWorker(queue *Queue, workerID string, jobTypes []string, pollInterval time.Duration, handler Handler, opts ...WorkerOption) *Worker {
	w := &Worker{
		queue:         queue,
		workerID:      workerID,
		jobTypes:      jobTypes,
		pollInterval:  pollInterval,
		handler:       handler,
		defaultLimits: DefaultLimits,
		notifier:      LogNotifier{},
	}
	for _, opt := range opts {
		opt(w)
	}
	return w
}

func (w *Worker) limitsFor(jobType string) Limits {
	if l, ok := w.jobTypeLimits[jobType]; ok {
		return l
	}
	return w.defaultLimits
}

// Run blockiert, bis ctx beendet wird, und verarbeitet dabei fortlaufend
// Jobs. Absturzsicherheit (Pruefung 1) entsteht NICHT durch Run selbst,
// sondern dadurch, dass ein abgestuerzter Prozess (der Run gar nicht mehr
// ausfuehrt) seine "processing"-Sperren nie verlaengert — ein ANDERER
// Worker-Prozess holt den Job nach Ablauf von staleLockAfter erneut ab
// (siehe Queue.Dequeue).
func (w *Worker) Run(ctx context.Context) {
	ticker := time.NewTicker(w.pollInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			w.processOne(ctx)
		}
	}
}

// processOne holt und verarbeitet EINEN Job, falls verfuegbar. Oeffentlich
// über RunOnce fuer Tests, die deterministisch (ohne Polling-Timing) einen
// einzelnen Verarbeitungsschritt auslösen wollen.
func (w *Worker) processOne(ctx context.Context) {
	job, err := w.queue.Dequeue(ctx, w.workerID, w.jobTypes)
	if err != nil {
		if !errors.Is(err, ErrNoJobAvailable) {
			log.Printf("jobqueue: dequeue fehlgeschlagen: %v", err)
		}
		return
	}

	// FDN-08 Akzeptanzkriterium 1: Job-ID dient Hintergrund-Jobs als
	// Korrelations-ID, analog zur X-Request-Id bei HTTP-Anfragen.
	ctx = logging.WithRequestID(ctx, job.ID)

	limits := w.limitsFor(job.JobType)

	// Groesse/Seitenzahl sind vorab aus den Job-Metadaten pruefbar (FDN-11,
	// Akzeptanzkriterium 1) — der Handler wird bei Ueberschreitung erst gar
	// nicht gestartet.
	if err := checkInputLimits(job.Payload, limits); err != nil {
		w.abortJob(ctx, job, err)
		return
	}

	handlerErr := w.runWithLimits(ctx, job, limits)
	if errors.Is(handlerErr, ErrLeaseLost) {
		log.Printf("jobqueue: sperre fuer job %q verloren, job wird abgebrochen und nicht veraendert", job.ID)
		return
	}
	if handlerErr != nil {
		var limitErr *LimitExceededError
		if errors.As(handlerErr, &limitErr) {
			w.abortJob(ctx, job, handlerErr)
			return
		}
		if err := w.queue.failOwned(ctx, job.ID, w.workerID, handlerErr); err != nil {
			log.Printf("jobqueue: fehlschlag fuer job %q nicht erfassbar: %v", job.ID, err)
		}
		return
	}
	if err := w.queue.completeOwned(ctx, job.ID, w.workerID); err != nil {
		log.Printf("jobqueue: abschluss fuer job %q fehlgeschlagen: %v", job.ID, err)
	}
}

// abortJob verschiebt einen Job wegen Ressourcenlimit-Verstoss in die
// Dead-Letter-Queue und benachrichtigt den Nutzer (Akzeptanzkriterium 2/3).
func (w *Worker) abortJob(ctx context.Context, job *Job, cause error) {
	if err := w.queue.failPermanentlyOwned(ctx, job.ID, w.workerID, cause); err != nil {
		log.Printf("jobqueue: abbruch fuer job %q nicht erfassbar: %v", job.ID, err)
	}
	if err := w.notifier.NotifyJobAborted(ctx, job, cause.Error()); err != nil {
		log.Printf("jobqueue: benachrichtigung fuer job %q fehlgeschlagen: %v", job.ID, err)
	}
}

// runWithLimits fuehrt den Handler mit Zeit- und Speicherobergrenze aus
// (Akzeptanzkriterium 1/2). Der Handler laeuft in einer eigenen Goroutine;
// bei Ueberschreitung kehrt runWithLimits SOFORT zurueck, statt auf ein
// Handler-Ende zu warten — genau das verhindert, dass ein einzelner
// entgleister Job den Worker blockiert (Akzeptanzkriterium 3). Respektiert
// der Handler den uebergebenen Kontext nicht, laeuft seine Goroutine im
// Hintergrund weiter aus — unvermeidbar ohne Sub-Prozess-Isolation, siehe
// "Nicht Bestandteil dieser Kachel".
func (w *Worker) runWithLimits(ctx context.Context, job *Job, limits Limits) error {
	jobCtx, lost := context.WithCancelCause(ctx)
	defer lost(nil)
	var cancel context.CancelFunc
	if limits.MaxDuration > 0 {
		jobCtx, cancel = context.WithTimeout(jobCtx, limits.MaxDuration)
		defer cancel()
	}
	defer w.holdLease(ctx, job.ID, lost)()

	done := make(chan error, 1)
	go func() {
		done <- w.handler(jobCtx, job)
	}()

	var memExceeded chan struct{}
	if limits.MaxMemoryBytes > 0 {
		memExceeded = make(chan struct{}, 1)
		stop := make(chan struct{})
		defer close(stop)
		go monitorMemory(limits.MaxMemoryBytes, memExceeded, stop)
	}

	select {
	case err := <-done:
		return err
	case <-jobCtx.Done():
		if errors.Is(context.Cause(jobCtx), ErrLeaseLost) {
			return ErrLeaseLost
		}
		if errors.Is(jobCtx.Err(), context.DeadlineExceeded) {
			return &LimitExceededError{Reason: fmt.Sprintf("verarbeitungszeit > %s", limits.MaxDuration)}
		}
		return jobCtx.Err()
	case <-memExceeded:
		if cancel != nil {
			cancel()
		}
		return &LimitExceededError{Reason: fmt.Sprintf("speicherverbrauch > %d bytes", limits.MaxMemoryBytes)}
	}
}

// monitorMemory ist eine prozessweite Annaeherung an ein Pro-Job-Speicherlimit:
// Go isoliert Speicher nicht pro Goroutine, daher wird der Heap-Zuwachs seit
// Jobstart (runtime.MemStats.Alloc) beobachtet. Bei Ueberschreitung wird
// GENAU EINMAL non-blocking signalisiert.
func monitorMemory(maxBytes uint64, exceeded chan<- struct{}, stop <-chan struct{}) {
	var start runtime.MemStats
	runtime.ReadMemStats(&start)

	ticker := time.NewTicker(memoryPollInterval)
	defer ticker.Stop()
	for {
		select {
		case <-stop:
			return
		case <-ticker.C:
			var cur runtime.MemStats
			runtime.ReadMemStats(&cur)
			if cur.Alloc > start.Alloc && cur.Alloc-start.Alloc > maxBytes {
				select {
				case exceeded <- struct{}{}:
				default:
				}
				return
			}
		}
	}
}

// RunOnce verarbeitet synchron genau einen Job (falls verfuegbar) und
// kehrt zurueck — fuer Tests, die ohne Polling-Intervall arbeiten wollen.
func (w *Worker) RunOnce(ctx context.Context) {
	w.processOne(ctx)
}